| /** @file |
| |
| A brief file description |
| |
| @section license License |
| |
| Licensed to the Apache Software Foundation (ASF) under one |
| or more contributor license agreements. See the NOTICE file |
| distributed with this work for additional information |
| regarding copyright ownership. The ASF licenses this file |
| to you under the Apache License, Version 2.0 (the |
| "License"); you may not use this file except in compliance |
| with the License. You may obtain a copy of the License at |
| |
| http://www.apache.org/licenses/LICENSE-2.0 |
| |
| Unless required by applicable law or agreed to in writing, software |
| distributed under the License is distributed on an "AS IS" BASIS, |
| WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
| See the License for the specific language governing permissions and |
| limitations under the License. |
| */ |
| |
| #include "P_Cache.h" |
| |
| #include "HttpCacheSM.h" //Added to get the scope of HttpCacheSM object. |
| |
| Action * |
| Cache::open_read(Continuation *cont, const CacheKey *key, CacheFragType type, const char *hostname, int host_len) |
| { |
| if (!CacheProcessor::IsCacheReady(type)) { |
| cont->handleEvent(CACHE_EVENT_OPEN_READ_FAILED, (void *)-ECACHE_NOT_READY); |
| return ACTION_RESULT_DONE; |
| } |
| ink_assert(caches[type] == this); |
| |
| Vol *vol = key_to_vol(key, hostname, host_len); |
| Dir result, *last_collision = nullptr; |
| ProxyMutex *mutex = cont->mutex.get(); |
| OpenDirEntry *od = nullptr; |
| CacheVC *c = nullptr; |
| { |
| CACHE_TRY_LOCK(lock, vol->mutex, mutex->thread_holding); |
| if (!lock.is_locked() || (od = vol->open_read(key)) || dir_probe(key, vol, &result, &last_collision)) { |
| c = new_CacheVC(cont); |
| SET_CONTINUATION_HANDLER(c, &CacheVC::openReadStartHead); |
| c->vio.op = VIO::READ; |
| c->base_stat = cache_read_active_stat; |
| CACHE_INCREMENT_DYN_STAT(c->base_stat + CACHE_STAT_ACTIVE); |
| c->first_key = c->key = c->earliest_key = *key; |
| c->vol = vol; |
| c->frag_type = type; |
| c->od = od; |
| } |
| if (!c) { |
| goto Lmiss; |
| } |
| if (!lock.is_locked()) { |
| CONT_SCHED_LOCK_RETRY(c); |
| return &c->_action; |
| } |
| if (c->od) { |
| goto Lwriter; |
| } |
| c->dir = result; |
| c->last_collision = last_collision; |
| switch (c->do_read_call(&c->key)) { |
| case EVENT_DONE: |
| return ACTION_RESULT_DONE; |
| case EVENT_RETURN: |
| goto Lcallreturn; |
| default: |
| return &c->_action; |
| } |
| } |
| Lmiss: |
| CACHE_INCREMENT_DYN_STAT(cache_read_failure_stat); |
| cont->handleEvent(CACHE_EVENT_OPEN_READ_FAILED, (void *)-ECACHE_NO_DOC); |
| return ACTION_RESULT_DONE; |
| Lwriter: |
| SET_CONTINUATION_HANDLER(c, &CacheVC::openReadFromWriter); |
| if (c->handleEvent(EVENT_IMMEDIATE, nullptr) == EVENT_DONE) { |
| return ACTION_RESULT_DONE; |
| } |
| return &c->_action; |
| Lcallreturn: |
| if (c->handleEvent(AIO_EVENT_DONE, nullptr) == EVENT_DONE) { |
| return ACTION_RESULT_DONE; |
| } |
| return &c->_action; |
| } |
| |
| Action * |
| Cache::open_read(Continuation *cont, const CacheKey *key, CacheHTTPHdr *request, const OverridableHttpConfigParams *params, |
| CacheFragType type, const char *hostname, int host_len) |
| { |
| if (!CacheProcessor::IsCacheReady(type)) { |
| cont->handleEvent(CACHE_EVENT_OPEN_READ_FAILED, (void *)-ECACHE_NOT_READY); |
| return ACTION_RESULT_DONE; |
| } |
| ink_assert(caches[type] == this); |
| |
| Vol *vol = key_to_vol(key, hostname, host_len); |
| Dir result, *last_collision = nullptr; |
| ProxyMutex *mutex = cont->mutex.get(); |
| OpenDirEntry *od = nullptr; |
| CacheVC *c = nullptr; |
| |
| { |
| CACHE_TRY_LOCK(lock, vol->mutex, mutex->thread_holding); |
| if (!lock.is_locked() || (od = vol->open_read(key)) || dir_probe(key, vol, &result, &last_collision)) { |
| c = new_CacheVC(cont); |
| c->first_key = c->key = c->earliest_key = *key; |
| c->vol = vol; |
| c->vio.op = VIO::READ; |
| c->base_stat = cache_read_active_stat; |
| CACHE_INCREMENT_DYN_STAT(c->base_stat + CACHE_STAT_ACTIVE); |
| c->request.copy_shallow(request); |
| c->frag_type = CACHE_FRAG_TYPE_HTTP; |
| c->params = params; |
| c->od = od; |
| } |
| if (!lock.is_locked()) { |
| SET_CONTINUATION_HANDLER(c, &CacheVC::openReadStartHead); |
| CONT_SCHED_LOCK_RETRY(c); |
| return &c->_action; |
| } |
| if (!c) { |
| goto Lmiss; |
| } |
| if (c->od) { |
| goto Lwriter; |
| } |
| // hit |
| c->dir = c->first_dir = result; |
| c->last_collision = last_collision; |
| SET_CONTINUATION_HANDLER(c, &CacheVC::openReadStartHead); |
| switch (c->do_read_call(&c->key)) { |
| case EVENT_DONE: |
| return ACTION_RESULT_DONE; |
| case EVENT_RETURN: |
| goto Lcallreturn; |
| default: |
| return &c->_action; |
| } |
| } |
| Lmiss: |
| CACHE_INCREMENT_DYN_STAT(cache_read_failure_stat); |
| cont->handleEvent(CACHE_EVENT_OPEN_READ_FAILED, (void *)-ECACHE_NO_DOC); |
| return ACTION_RESULT_DONE; |
| Lwriter: |
| // this is a horrible violation of the interface and should be fixed (FIXME) |
| ((HttpCacheSM *)cont)->set_readwhilewrite_inprogress(true); |
| SET_CONTINUATION_HANDLER(c, &CacheVC::openReadFromWriter); |
| if (c->handleEvent(EVENT_IMMEDIATE, nullptr) == EVENT_DONE) { |
| return ACTION_RESULT_DONE; |
| } |
| return &c->_action; |
| Lcallreturn: |
| if (c->handleEvent(AIO_EVENT_DONE, nullptr) == EVENT_DONE) { |
| return ACTION_RESULT_DONE; |
| } |
| return &c->_action; |
| } |
| |
| uint32_t |
| CacheVC::load_http_info(CacheHTTPInfoVector *info, Doc *doc, RefCountObj *block_ptr) |
| { |
| uint32_t zret = info->get_handles(doc->hdr(), doc->hlen, block_ptr); |
| if (!this->f.doc_from_ram_cache && // ram cache is always already fixed up. |
| // If this is an old object, the object version will be old or 0, in either case this is |
| // correct. Forget the 4.2 compatibility, always update older versioned objects. |
| ts::VersionNumber(doc->v_major, doc->v_minor) < CACHE_DB_VERSION) { |
| for (int i = info->xcount - 1; i >= 0; --i) { |
| info->data(i).alternate.m_alt->m_response_hdr.m_mime->recompute_accelerators_and_presence_bits(); |
| info->data(i).alternate.m_alt->m_request_hdr.m_mime->recompute_accelerators_and_presence_bits(); |
| } |
| } |
| return zret; |
| } |
| |
| int |
| CacheVC::openReadFromWriterFailure(int event, Event *e) |
| { |
| od = nullptr; |
| vector.clear(false); |
| CACHE_INCREMENT_DYN_STAT(cache_read_failure_stat); |
| CACHE_INCREMENT_DYN_STAT(cache_read_busy_failure_stat); |
| _action.continuation->handleEvent(event, e); |
| free_CacheVC(this); |
| return EVENT_DONE; |
| } |
| |
| int |
| CacheVC::openReadChooseWriter(int /* event ATS_UNUSED */, Event * /* e ATS_UNUSED */) |
| { |
| intptr_t err = ECACHE_DOC_BUSY; |
| CacheVC *w = nullptr; |
| |
| ink_assert(vol->mutex->thread_holding == mutex->thread_holding && write_vc == nullptr); |
| |
| if (!od) { |
| return EVENT_RETURN; |
| } |
| |
| if (frag_type != CACHE_FRAG_TYPE_HTTP) { |
| ink_assert(od->num_writers == 1); |
| w = od->writers.head; |
| if (w->start_time > start_time || w->closed < 0) { |
| od = nullptr; |
| return EVENT_RETURN; |
| } |
| if (!w->closed) { |
| return -err; |
| } |
| write_vc = w; |
| } else { |
| write_vector = &od->vector; |
| int write_vec_cnt = write_vector->count(); |
| for (int c = 0; c < write_vec_cnt; c++) { |
| vector.insert(write_vector->get(c)); |
| } |
| // check if all the writers who came before this reader have |
| // set the http_info. |
| for (w = static_cast<CacheVC *>(od->writers.head); w; w = static_cast<CacheVC *>(w->opendir_link.next)) { |
| if (w->start_time > start_time || w->closed < 0) { |
| continue; |
| } |
| if (!w->closed && !cache_config_read_while_writer) { |
| return -err; |
| } |
| if (w->alternate_index != CACHE_ALT_INDEX_DEFAULT) { |
| continue; |
| } |
| |
| if (!w->closed && !w->alternate.valid()) { |
| od = nullptr; |
| ink_assert(!write_vc); |
| vector.clear(false); |
| return EVENT_CONT; |
| } |
| // construct the vector from the writers. |
| int alt_ndx = CACHE_ALT_INDEX_DEFAULT; |
| if (w->f.update) { |
| // all Update cases. Need to get the alternate index. |
| alt_ndx = get_alternate_index(&vector, w->update_key); |
| // if its an alternate delete |
| if (!w->alternate.valid()) { |
| if (alt_ndx >= 0) { |
| vector.remove(alt_ndx, false); |
| } |
| continue; |
| } |
| } |
| if (w->alternate.valid()) { |
| vector.insert(&w->alternate, alt_ndx); |
| } |
| } |
| |
| if (!vector.count()) { |
| if (od->reading_vec) { |
| // the writer(s) are reading the vector, so there is probably |
| // an old vector. Since this reader came before any of the |
| // current writers, we should return the old data |
| od = nullptr; |
| return EVENT_RETURN; |
| } |
| return -ECACHE_NO_DOC; |
| } |
| if (cache_config_select_alternate) { |
| alternate_index = HttpTransactCache::SelectFromAlternates(&vector, &request, params); |
| if (alternate_index < 0) { |
| return -ECACHE_ALT_MISS; |
| } |
| } else { |
| alternate_index = 0; |
| } |
| CacheHTTPInfo *obj = vector.get(alternate_index); |
| for (w = static_cast<CacheVC *>(od->writers.head); w; w = static_cast<CacheVC *>(w->opendir_link.next)) { |
| if (obj->m_alt == w->alternate.m_alt) { |
| write_vc = w; |
| break; |
| } |
| } |
| vector.clear(false); |
| if (!write_vc) { |
| DDebug("cache_read_agg", "%p: key: %X writer alternate different: %d", this, first_key.slice32(1), alternate_index); |
| od = nullptr; |
| return EVENT_RETURN; |
| } |
| |
| DDebug("cache_read_agg", "%p: key: %X eKey: %d # alts: %d, ndx: %d, # writers: %d writer: %p", this, first_key.slice32(1), |
| write_vc->earliest_key.slice32(1), vector.count(), alternate_index, od->num_writers, write_vc); |
| } |
| return EVENT_NONE; |
| } |
| |
| int |
| CacheVC::openReadFromWriter(int event, Event *e) |
| { |
| if (!f.read_from_writer_called) { |
| // The assignment to last_collision as nullptr was |
| // made conditional after INKqa08411 |
| last_collision = nullptr; |
| // Let's restart the clock from here - the first time this a reader |
| // gets in this state. Its possible that the open_read was called |
| // before the open_write, but the reader could not get the volume |
| // lock. If we don't reset the clock here, we won't choose any writer |
| // and hence fail the read request. |
| start_time = Thread::get_hrtime(); |
| f.read_from_writer_called = 1; |
| } |
| cancel_trigger(); |
| intptr_t err = ECACHE_DOC_BUSY; |
| DDebug("cache_read_agg", "%p: key: %X In openReadFromWriter", this, first_key.slice32(1)); |
| if (_action.cancelled) { |
| od = nullptr; // only open for read so no need to close |
| return free_CacheVC(this); |
| } |
| CACHE_TRY_LOCK(lock, vol->mutex, mutex->thread_holding); |
| if (!lock.is_locked()) { |
| VC_SCHED_LOCK_RETRY(); |
| } |
| od = vol->open_read(&first_key); // recheck in case the lock failed |
| if (!od) { |
| MUTEX_RELEASE(lock); |
| write_vc = nullptr; |
| SET_HANDLER(&CacheVC::openReadStartHead); |
| return openReadStartHead(event, e); |
| } else { |
| ink_assert(od == vol->open_read(&first_key)); |
| } |
| if (!write_vc) { |
| int ret = openReadChooseWriter(event, e); |
| if (ret < 0) { |
| MUTEX_RELEASE(lock); |
| SET_HANDLER(&CacheVC::openReadFromWriterFailure); |
| return openReadFromWriterFailure(CACHE_EVENT_OPEN_READ_FAILED, reinterpret_cast<Event *>(ret)); |
| } else if (ret == EVENT_RETURN) { |
| MUTEX_RELEASE(lock); |
| SET_HANDLER(&CacheVC::openReadStartHead); |
| return openReadStartHead(event, e); |
| } else if (ret == EVENT_CONT) { |
| ink_assert(!write_vc); |
| if (writer_lock_retry < cache_config_read_while_writer_max_retries) { |
| VC_SCHED_WRITER_RETRY(); |
| } else { |
| return openReadFromWriterFailure(CACHE_EVENT_OPEN_READ_FAILED, (Event *)-err); |
| } |
| } else { |
| ink_assert(write_vc); |
| } |
| } else { |
| if (writer_done()) { |
| MUTEX_RELEASE(lock); |
| DDebug("cache_read_agg", "%p: key: %X writer %p has left, continuing as normal read", this, first_key.slice32(1), write_vc); |
| od = nullptr; |
| write_vc = nullptr; |
| SET_HANDLER(&CacheVC::openReadStartHead); |
| return openReadStartHead(event, e); |
| } |
| } |
| OpenDirEntry *cod = od; |
| od = nullptr; |
| // someone is currently writing the document |
| if (write_vc->closed < 0) { |
| MUTEX_RELEASE(lock); |
| write_vc = nullptr; |
| // writer aborted, continue as if there is no writer |
| SET_HANDLER(&CacheVC::openReadStartHead); |
| return openReadStartHead(EVENT_IMMEDIATE, nullptr); |
| } |
| // allow reading from unclosed writer for http requests only. |
| ink_assert(frag_type == CACHE_FRAG_TYPE_HTTP || write_vc->closed); |
| if (!write_vc->closed && !write_vc->fragment) { |
| if (!cache_config_read_while_writer || frag_type != CACHE_FRAG_TYPE_HTTP || |
| writer_lock_retry >= cache_config_read_while_writer_max_retries) { |
| MUTEX_RELEASE(lock); |
| return openReadFromWriterFailure(CACHE_EVENT_OPEN_READ_FAILED, (Event *)-err); |
| } |
| DDebug("cache_read_agg", "%p: key: %X writer: closed:%d, fragment:%d, retry: %d", this, first_key.slice32(1), write_vc->closed, |
| write_vc->fragment, writer_lock_retry); |
| VC_SCHED_WRITER_RETRY(); |
| } |
| |
| CACHE_TRY_LOCK(writer_lock, write_vc->mutex, mutex->thread_holding); |
| if (!writer_lock.is_locked()) { |
| DDebug("cache_read_agg", "%p: key: %X lock miss", this, first_key.slice32(1)); |
| VC_SCHED_LOCK_RETRY(); |
| } |
| MUTEX_RELEASE(lock); |
| |
| if (!write_vc->io.ok()) { |
| return openReadFromWriterFailure(CACHE_EVENT_OPEN_READ_FAILED, (Event *)-err); |
| } |
| if (frag_type == CACHE_FRAG_TYPE_HTTP) { |
| DDebug("cache_read_agg", "%p: key: %X http passed stage 1, closed: %d, frag: %d", this, first_key.slice32(1), write_vc->closed, |
| write_vc->fragment); |
| if (!write_vc->alternate.valid()) { |
| return openReadFromWriterFailure(CACHE_EVENT_OPEN_READ_FAILED, (Event *)-err); |
| } |
| alternate.copy(&write_vc->alternate); |
| vector.insert(&alternate); |
| alternate.object_key_get(&key); |
| write_vc->f.readers = 1; |
| if (!(write_vc->f.update && write_vc->total_len == 0)) { |
| key = write_vc->earliest_key; |
| if (!write_vc->closed) { |
| alternate.object_size_set(write_vc->vio.nbytes); |
| } else { |
| alternate.object_size_set(write_vc->total_len); |
| } |
| } else { |
| key = write_vc->update_key; |
| ink_assert(write_vc->closed); |
| DDebug("cache_read_agg", "%p: key: %X writer header update", this, first_key.slice32(1)); |
| // Update case (b) : grab doc_len from the writer's alternate |
| doc_len = alternate.object_size_get(); |
| if (write_vc->update_key == cod->single_doc_key && (cod->move_resident_alt || write_vc->f.rewrite_resident_alt) && |
| write_vc->first_buf.get()) { |
| // the resident alternate is being updated and its a |
| // header only update. The first_buf of the writer has the |
| // document body. |
| Doc *doc = reinterpret_cast<Doc *>(write_vc->first_buf->data()); |
| writer_buf = new_IOBufferBlock(write_vc->first_buf, doc->data_len(), doc->prefix_len()); |
| MUTEX_RELEASE(writer_lock); |
| ink_assert(doc_len == doc->data_len()); |
| length = doc_len; |
| f.single_fragment = 1; |
| doc_pos = 0; |
| earliest_key = key; |
| dir_clean(&first_dir); |
| dir_clean(&earliest_dir); |
| SET_HANDLER(&CacheVC::openReadFromWriterMain); |
| CACHE_INCREMENT_DYN_STAT(cache_read_busy_success_stat); |
| return callcont(CACHE_EVENT_OPEN_READ); |
| } |
| // want to snarf the new headers from the writer |
| // and then continue as if nothing happened |
| last_collision = nullptr; |
| MUTEX_RELEASE(writer_lock); |
| SET_HANDLER(&CacheVC::openReadStartEarliest); |
| return openReadStartEarliest(event, e); |
| } |
| } else { |
| DDebug("cache_read_agg", "%p: key: %X non-http passed stage 1", this, first_key.slice32(1)); |
| key = write_vc->earliest_key; |
| } |
| if (write_vc->fragment) { |
| doc_len = write_vc->vio.nbytes; |
| last_collision = nullptr; |
| DDebug("cache_read_agg", "%p: key: %X closed: %d, fragment: %d, len: %d starting first fragment", this, first_key.slice32(1), |
| write_vc->closed, write_vc->fragment, (int)doc_len); |
| MUTEX_RELEASE(writer_lock); |
| // either a header + body update or a new document |
| SET_HANDLER(&CacheVC::openReadStartEarliest); |
| return openReadStartEarliest(event, e); |
| } |
| writer_buf = write_vc->blocks; |
| writer_offset = write_vc->offset; |
| length = write_vc->length; |
| // copy the vector |
| f.single_fragment = !write_vc->fragment; // single fragment doc |
| doc_pos = 0; |
| earliest_key = write_vc->earliest_key; |
| ink_assert(earliest_key == key); |
| doc_len = write_vc->total_len; |
| dir_clean(&first_dir); |
| dir_clean(&earliest_dir); |
| DDebug("cache_read_agg", "%p: key: %X %X: single fragment read", this, first_key.slice32(1), key.slice32(0)); |
| MUTEX_RELEASE(writer_lock); |
| SET_HANDLER(&CacheVC::openReadFromWriterMain); |
| CACHE_INCREMENT_DYN_STAT(cache_read_busy_success_stat); |
| return callcont(CACHE_EVENT_OPEN_READ); |
| } |
| |
| int |
| CacheVC::openReadFromWriterMain(int /* event ATS_UNUSED */, Event * /* e ATS_UNUSED */) |
| { |
| cancel_trigger(); |
| if (seek_to) { |
| vio.ndone = seek_to; |
| seek_to = 0; |
| } |
| IOBufferBlock *b = nullptr; |
| int64_t ntodo = vio.ntodo(); |
| if (ntodo <= 0) { |
| return EVENT_CONT; |
| } |
| if (length < (static_cast<int64_t>(doc_len)) - vio.ndone) { |
| DDebug("cache_read_agg", "truncation %X", first_key.slice32(1)); |
| if (is_action_tag_set("cache")) { |
| ink_release_assert(false); |
| } |
| Warning("Document %X truncated at %d of %d, reading from writer", first_key.slice32(1), (int)vio.ndone, (int)doc_len); |
| return calluser(VC_EVENT_ERROR); |
| } |
| /* its possible that the user did a do_io_close before |
| openWriteWriteDone was called. */ |
| if (length > (static_cast<int64_t>(doc_len)) - vio.ndone) { |
| int64_t skip_bytes = length - (doc_len - vio.ndone); |
| iobufferblock_skip(writer_buf.get(), &writer_offset, &length, skip_bytes); |
| } |
| int64_t bytes = length; |
| if (bytes > vio.ntodo()) { |
| bytes = vio.ntodo(); |
| } |
| if (vio.ndone >= static_cast<int64_t>(doc_len)) { |
| ink_assert(bytes <= 0); |
| // reached the end of the document and the user still wants more |
| return calluser(VC_EVENT_EOS); |
| } |
| b = iobufferblock_clone(writer_buf.get(), writer_offset, bytes); |
| writer_buf = iobufferblock_skip(writer_buf.get(), &writer_offset, &length, bytes); |
| vio.buffer.writer()->append_block(b); |
| vio.ndone += bytes; |
| if (vio.ntodo() <= 0) { |
| return calluser(VC_EVENT_READ_COMPLETE); |
| } else { |
| return calluser(VC_EVENT_READ_READY); |
| } |
| } |
| |
| int |
| CacheVC::openReadClose(int event, Event * /* e ATS_UNUSED */) |
| { |
| cancel_trigger(); |
| if (is_io_in_progress()) { |
| if (event != AIO_EVENT_DONE) { |
| return EVENT_CONT; |
| } |
| set_io_not_in_progress(); |
| } |
| CACHE_TRY_LOCK(lock, vol->mutex, mutex->thread_holding); |
| if (!lock.is_locked()) { |
| VC_SCHED_LOCK_RETRY(); |
| } |
| if (f.hit_evacuate && dir_valid(vol, &first_dir) && closed > 0) { |
| if (f.single_fragment) { |
| vol->force_evacuate_head(&first_dir, dir_pinned(&first_dir)); |
| } else if (dir_valid(vol, &earliest_dir)) { |
| vol->force_evacuate_head(&first_dir, dir_pinned(&first_dir)); |
| vol->force_evacuate_head(&earliest_dir, dir_pinned(&earliest_dir)); |
| } |
| } |
| vol->close_read(this); |
| return free_CacheVC(this); |
| } |
| |
| int |
| CacheVC::openReadReadDone(int event, Event *e) |
| { |
| Doc *doc = nullptr; |
| |
| cancel_trigger(); |
| if (event == EVENT_IMMEDIATE) { |
| return EVENT_CONT; |
| } |
| set_io_not_in_progress(); |
| { |
| CACHE_TRY_LOCK(lock, vol->mutex, mutex->thread_holding); |
| if (!lock.is_locked()) { |
| VC_SCHED_LOCK_RETRY(); |
| } |
| if (event == AIO_EVENT_DONE && !io.ok()) { |
| goto Lerror; |
| } |
| if (last_collision && // no missed lock |
| dir_valid(vol, &dir)) // object still valid |
| { |
| doc = reinterpret_cast<Doc *>(buf->data()); |
| if (doc->magic != DOC_MAGIC) { |
| char tmpstring[CRYPTO_HEX_SIZE]; |
| if (doc->magic == DOC_CORRUPT) { |
| Warning("Middle: Doc checksum does not match for %s", key.toHexStr(tmpstring)); |
| } else { |
| Warning("Middle: Doc magic does not match for %s", key.toHexStr(tmpstring)); |
| } |
| goto Lerror; |
| } |
| if (doc->key == key) { |
| goto LreadMain; |
| } |
| } |
| if (last_collision && dir_offset(&dir) != dir_offset(last_collision)) { |
| last_collision = nullptr; // object has been/is being overwritten |
| } |
| if (dir_probe(&key, vol, &dir, &last_collision)) { |
| int ret = do_read_call(&key); |
| if (ret == EVENT_RETURN) { |
| goto Lcallreturn; |
| } |
| return EVENT_CONT; |
| } else if (write_vc) { |
| if (writer_done()) { |
| last_collision = nullptr; |
| while (dir_probe(&earliest_key, vol, &dir, &last_collision)) { |
| if (dir_offset(&dir) == dir_offset(&earliest_dir)) { |
| DDebug("cache_read_agg", "%p: key: %X ReadRead complete: %d", this, first_key.slice32(1), (int)vio.ndone); |
| doc_len = vio.ndone; |
| goto Ldone; |
| } |
| } |
| DDebug("cache_read_agg", "%p: key: %X ReadRead writer aborted: %d", this, first_key.slice32(1), (int)vio.ndone); |
| goto Lerror; |
| } |
| if (writer_lock_retry < cache_config_read_while_writer_max_retries) { |
| DDebug("cache_read_agg", "%p: key: %X ReadRead retrying: %d", this, first_key.slice32(1), (int)vio.ndone); |
| VC_SCHED_WRITER_RETRY(); // wait for writer |
| } else { |
| DDebug("cache_read_agg", "%p: key: %X ReadRead retries exhausted, bailing..: %d", this, first_key.slice32(1), |
| (int)vio.ndone); |
| goto Ldone; |
| } |
| } |
| // fall through for truncated documents |
| } |
| Lerror : { |
| char tmpstring[CRYPTO_HEX_SIZE]; |
| if (request.valid()) { |
| int url_length; |
| const char *url_text = request.url_get()->string_get_ref(&url_length); |
| Warning("Document %s truncated, url[%.*s] .. clearing", earliest_key.toHexStr(tmpstring), url_length, url_text); |
| } else { |
| Warning("Document %s truncated .. clearing", earliest_key.toHexStr(tmpstring)); |
| } |
| dir_delete(&earliest_key, vol, &earliest_dir); |
| return calluser(VC_EVENT_ERROR); |
| } |
| Ldone: |
| return calluser(VC_EVENT_EOS); |
| Lcallreturn: |
| return handleEvent(AIO_EVENT_DONE, nullptr); |
| LreadMain: |
| fragment++; |
| doc_pos = doc->prefix_len(); |
| next_CacheKey(&key, &key); |
| SET_HANDLER(&CacheVC::openReadMain); |
| return openReadMain(event, e); |
| } |
| |
| int |
| CacheVC::openReadMain(int /* event ATS_UNUSED */, Event * /* e ATS_UNUSED */) |
| { |
| cancel_trigger(); |
| Doc *doc = reinterpret_cast<Doc *>(buf->data()); |
| int64_t ntodo = vio.ntodo(); |
| int64_t bytes = doc->len - doc_pos; |
| IOBufferBlock *b = nullptr; |
| if (seek_to) { // handle do_io_pread |
| if (seek_to >= doc_len) { |
| vio.ndone = doc_len; |
| return calluser(VC_EVENT_EOS); |
| } |
| HTTPInfo::FragOffset *frags = alternate.get_frag_table(); |
| if (is_debug_tag_set("cache_seek")) { |
| char b[CRYPTO_HEX_SIZE], c[CRYPTO_HEX_SIZE]; |
| Debug("cache_seek", "Seek @ %" PRId64 " in %s from #%d @ %" PRId64 "/%d:%s", seek_to, first_key.toHexStr(b), fragment, |
| doc_pos, doc->len, doc->key.toHexStr(c)); |
| } |
| /* Because single fragment objects can migrate to hang off an alt vector |
| they can appear to the VC as multi-fragment when they are not really. |
| The essential difference is the existence of a fragment table. |
| */ |
| if (frags) { |
| int target = 0; |
| HTTPInfo::FragOffset next_off = frags[target]; |
| int lfi = static_cast<int>(alternate.get_frag_offset_count()) - 1; |
| ink_assert(lfi >= 0); // because it's not a single frag doc. |
| |
| /* Note: frag[i].offset is the offset of the first byte past the |
| i'th fragment. So frag[0].offset is the offset of the first |
| byte of fragment 1. In addition the # of fragments is one |
| more than the fragment table length, the start of the last |
| fragment being the last offset in the table. |
| */ |
| if (fragment == 0 || seek_to < frags[fragment - 1] || (fragment <= lfi && frags[fragment] <= seek_to)) { |
| // search from frag 0 on to find the proper frag |
| while (seek_to >= next_off && target < lfi) { |
| next_off = frags[++target]; |
| } |
| if (target == lfi && seek_to >= next_off) { |
| ++target; |
| } |
| } else { // shortcut if we are in the fragment already |
| target = fragment; |
| } |
| if (target != fragment) { |
| // Lread will read the next fragment always, so if that |
| // is the one we want, we don't need to do anything |
| int cfi = fragment; |
| --target; |
| while (target > fragment) { |
| next_CacheKey(&key, &key); |
| ++fragment; |
| } |
| while (target < fragment) { |
| prev_CacheKey(&key, &key); |
| --fragment; |
| } |
| |
| if (is_debug_tag_set("cache_seek")) { |
| char target_key_str[CRYPTO_HEX_SIZE]; |
| key.toHexStr(target_key_str); |
| Debug("cache_seek", "Seek #%d @ %" PRId64 " -> #%d @ %" PRId64 ":%s", cfi, doc_pos, target, seek_to, target_key_str); |
| } |
| goto Lread; |
| } |
| } |
| doc_pos = doc->prefix_len() + seek_to; |
| if (fragment && frags) { |
| doc_pos -= static_cast<int64_t>(frags[fragment - 1]); |
| } |
| vio.ndone = 0; |
| seek_to = 0; |
| ntodo = vio.ntodo(); |
| bytes = doc->len - doc_pos; |
| if (is_debug_tag_set("cache_seek")) { |
| char target_key_str[CRYPTO_HEX_SIZE]; |
| Debug("cache_seek", "Read # %d @ %" PRId64 "/%d for %" PRId64 " %s", fragment, doc_pos, doc->len, bytes, |
| key.toHexStr(target_key_str)); |
| } |
| |
| // This shouldn't happen for HTTP assets but it does |
| // occasionally in production. This is a temporary fix |
| // to clean up broken objects until the root cause can |
| // be found. It must be the case that either the fragment |
| // offsets are incorrect or a fragment table isn't being |
| // created when it should be. |
| if (frag_type == CACHE_FRAG_TYPE_HTTP && bytes < 0) { |
| char xt[CRYPTO_HEX_SIZE]; |
| char yt[CRYPTO_HEX_SIZE]; |
| |
| int url_length = 0; |
| char const *url_text = nullptr; |
| if (request.valid()) { |
| url_text = request.url_get()->string_get_ref(&url_length); |
| } |
| |
| int64_t prev_frag_size = 0; |
| if (fragment && frags) { |
| prev_frag_size = static_cast<int64_t>(frags[fragment - 1]); |
| } |
| |
| Warning("cache_seek range request bug: read %s targ %s - %s frag # %d (prev_frag %" PRId64 ") @ %" PRId64 "/%d for %" PRId64 |
| " tot %" PRId64 " url '%.*s'", |
| doc->key.toHexStr(xt), key.toHexStr(yt), f.single_fragment ? "single" : "multi", fragment, prev_frag_size, doc_pos, |
| doc->len, bytes, doc->total_len, url_length, url_text); |
| |
| doc->magic = DOC_CORRUPT; |
| |
| CACHE_INCREMENT_DYN_STAT(cache_read_seek_fail_stat); |
| |
| CACHE_TRY_LOCK(lock, vol->mutex, mutex->thread_holding); |
| if (!lock.is_locked()) { |
| SET_HANDLER(&CacheVC::openReadDirDelete); |
| VC_SCHED_LOCK_RETRY(); |
| } |
| |
| dir_delete(&earliest_key, vol, &earliest_dir); |
| goto Lerror; |
| } |
| } |
| if (ntodo <= 0) { |
| return EVENT_CONT; |
| } |
| if (vio.buffer.writer()->max_read_avail() > vio.buffer.writer()->water_mark && vio.ndone) { // initiate read of first block |
| return EVENT_CONT; |
| } |
| if ((bytes <= 0) && vio.ntodo() >= 0) { |
| goto Lread; |
| } |
| if (bytes > vio.ntodo()) { |
| bytes = vio.ntodo(); |
| } |
| b = new_IOBufferBlock(buf, bytes, doc_pos); |
| b->_buf_end = b->_end; |
| vio.buffer.writer()->append_block(b); |
| vio.ndone += bytes; |
| doc_pos += bytes; |
| if (vio.ntodo() <= 0) { |
| return calluser(VC_EVENT_READ_COMPLETE); |
| } else { |
| if (calluser(VC_EVENT_READ_READY) == EVENT_DONE) { |
| return EVENT_DONE; |
| } |
| // we have to keep reading until we give the user all the |
| // bytes it wanted or we hit the watermark. |
| if (vio.ntodo() > 0 && !vio.buffer.writer()->high_water()) { |
| goto Lread; |
| } |
| return EVENT_CONT; |
| } |
| Lread : { |
| if (vio.ndone >= static_cast<int64_t>(doc_len)) { |
| // reached the end of the document and the user still wants more |
| return calluser(VC_EVENT_EOS); |
| } |
| last_collision = nullptr; |
| writer_lock_retry = 0; |
| // if the state machine calls reenable on the callback from the cache, |
| // we set up a schedule_imm event. The openReadReadDone discards |
| // EVENT_IMMEDIATE events. So, we have to cancel that trigger and set |
| // a new EVENT_INTERVAL event. |
| cancel_trigger(); |
| CACHE_TRY_LOCK(lock, vol->mutex, mutex->thread_holding); |
| if (!lock.is_locked()) { |
| SET_HANDLER(&CacheVC::openReadMain); |
| VC_SCHED_LOCK_RETRY(); |
| } |
| if (dir_probe(&key, vol, &dir, &last_collision)) { |
| SET_HANDLER(&CacheVC::openReadReadDone); |
| int ret = do_read_call(&key); |
| if (ret == EVENT_RETURN) { |
| goto Lcallreturn; |
| } |
| return EVENT_CONT; |
| } else if (write_vc) { |
| if (writer_done()) { |
| last_collision = nullptr; |
| while (dir_probe(&earliest_key, vol, &dir, &last_collision)) { |
| if (dir_offset(&dir) == dir_offset(&earliest_dir)) { |
| DDebug("cache_read_agg", "%p: key: %X ReadMain complete: %d", this, first_key.slice32(1), (int)vio.ndone); |
| doc_len = vio.ndone; |
| goto Leos; |
| } |
| } |
| DDebug("cache_read_agg", "%p: key: %X ReadMain writer aborted: %d", this, first_key.slice32(1), (int)vio.ndone); |
| goto Lerror; |
| } |
| DDebug("cache_read_agg", "%p: key: %X ReadMain retrying: %d", this, first_key.slice32(1), (int)vio.ndone); |
| SET_HANDLER(&CacheVC::openReadMain); |
| VC_SCHED_WRITER_RETRY(); |
| } |
| if (is_action_tag_set("cache")) { |
| ink_release_assert(false); |
| } |
| Warning("Document %X truncated at %d of %d, missing fragment %X", first_key.slice32(1), (int)vio.ndone, (int)doc_len, |
| key.slice32(1)); |
| // remove the directory entry |
| dir_delete(&earliest_key, vol, &earliest_dir); |
| } |
| Lerror: |
| return calluser(VC_EVENT_ERROR); |
| Leos: |
| return calluser(VC_EVENT_EOS); |
| Lcallreturn: |
| return handleEvent(AIO_EVENT_DONE, nullptr); |
| } |
| |
| /* |
| This code follows CacheVC::openReadStartHead closely, |
| if you change this you might have to change that. |
| */ |
| int |
| CacheVC::openReadStartEarliest(int /* event ATS_UNUSED */, Event * /* e ATS_UNUSED */) |
| { |
| int ret = 0; |
| Doc *doc = nullptr; |
| cancel_trigger(); |
| set_io_not_in_progress(); |
| if (_action.cancelled) { |
| return free_CacheVC(this); |
| } |
| { |
| CACHE_TRY_LOCK(lock, vol->mutex, mutex->thread_holding); |
| if (!lock.is_locked()) { |
| VC_SCHED_LOCK_RETRY(); |
| } |
| if (!buf) { |
| goto Lread; |
| } |
| if (!io.ok()) { |
| goto Ldone; |
| } |
| // an object needs to be outside the aggregation window in order to be |
| // be evacuated as it is read |
| if (!dir_agg_valid(vol, &dir)) { |
| // a directory entry which is no longer valid may have been overwritten |
| if (!dir_valid(vol, &dir)) { |
| last_collision = nullptr; |
| } |
| goto Lread; |
| } |
| doc = reinterpret_cast<Doc *>(buf->data()); |
| if (doc->magic != DOC_MAGIC) { |
| char tmpstring[CRYPTO_HEX_SIZE]; |
| if (is_action_tag_set("cache")) { |
| ink_release_assert(false); |
| } |
| if (doc->magic == DOC_CORRUPT) { |
| Warning("Earliest: Doc checksum does not match for %s", key.toHexStr(tmpstring)); |
| } else { |
| Warning("Earliest : Doc magic does not match for %s", key.toHexStr(tmpstring)); |
| } |
| // remove the dir entry |
| dir_delete(&key, vol, &dir); |
| // try going through the directory entries again |
| // in case the dir entry we deleted doesnt correspond |
| // to the key we are looking for. This is possible |
| // because of directory collisions |
| last_collision = nullptr; |
| goto Lread; |
| } |
| if (!(doc->key == key)) { // collision |
| goto Lread; |
| } |
| // success |
| earliest_key = key; |
| doc_pos = doc->prefix_len(); |
| next_CacheKey(&key, &doc->key); |
| vol->begin_read(this); |
| if (vol->within_hit_evacuate_window(&earliest_dir) && |
| (!cache_config_hit_evacuate_size_limit || doc_len <= static_cast<uint64_t>(cache_config_hit_evacuate_size_limit))) { |
| DDebug("cache_hit_evac", "dir: %" PRId64 ", write: %" PRId64 ", phase: %d", dir_offset(&earliest_dir), |
| vol->offset_to_vol_offset(vol->header->write_pos), vol->header->phase); |
| f.hit_evacuate = 1; |
| } |
| goto Lsuccess; |
| Lread: |
| if (dir_probe(&key, vol, &earliest_dir, &last_collision) || dir_lookaside_probe(&key, vol, &earliest_dir, nullptr)) { |
| dir = earliest_dir; |
| if ((ret = do_read_call(&key)) == EVENT_RETURN) { |
| goto Lcallreturn; |
| } |
| return ret; |
| } |
| // read has detected that alternate does not exist in the cache. |
| // rewrite the vector. |
| if (!f.read_from_writer_called && frag_type == CACHE_FRAG_TYPE_HTTP) { |
| // don't want any writers while we are evacuating the vector |
| if (!vol->open_write(this, false, 1)) { |
| Doc *doc1 = reinterpret_cast<Doc *>(first_buf->data()); |
| uint32_t len = this->load_http_info(write_vector, doc1); |
| ink_assert(len == doc1->hlen && write_vector->count() > 0); |
| write_vector->remove(alternate_index, true); |
| // if the vector had one alternate, delete it's directory entry |
| if (len != doc1->hlen || !write_vector->count()) { |
| // sometimes the delete fails when there is a race and another read |
| // finds that the directory entry has been overwritten |
| // (cannot assert on the return value) |
| dir_delete(&first_key, vol, &first_dir); |
| } else { |
| buf = nullptr; |
| last_collision = nullptr; |
| write_len = 0; |
| header_len = write_vector->marshal_length(); |
| f.evac_vector = 1; |
| f.use_first_key = 1; |
| key = first_key; |
| // always use od->first_dir to overwrite a directory. |
| // If an evacuation happens while a vector is being updated |
| // the evacuator changes the od->first_dir to the new directory |
| // that it inserted |
| od->first_dir = first_dir; |
| od->writing_vec = true; |
| earliest_key = zero_key; |
| |
| // set up this VC as a alternate delete write_vc |
| vio.op = VIO::WRITE; |
| total_len = 0; |
| f.update = 1; |
| alternate_index = CACHE_ALT_REMOVED; |
| ///////////////////////////////////////////////////////////////// |
| // change to create a directory entry for a resident alternate // |
| // when another alternate does not exist. // |
| ///////////////////////////////////////////////////////////////// |
| if (doc1->total_len > 0) { |
| od->move_resident_alt = true; |
| od->single_doc_key = doc1->key; |
| dir_assign(&od->single_doc_dir, &dir); |
| dir_set_tag(&od->single_doc_dir, od->single_doc_key.slice32(2)); |
| } |
| SET_HANDLER(&CacheVC::openReadVecWrite); |
| if ((ret = do_write_call()) == EVENT_RETURN) { |
| goto Lcallreturn; |
| } |
| return ret; |
| } |
| } |
| } |
| // open write failure - another writer, so don't modify the vector |
| Ldone: |
| if (od) { |
| vol->close_write(this); |
| } |
| } |
| CACHE_INCREMENT_DYN_STAT(cache_read_failure_stat); |
| _action.continuation->handleEvent(CACHE_EVENT_OPEN_READ_FAILED, (void *)-ECACHE_NO_DOC); |
| return free_CacheVC(this); |
| Lcallreturn: |
| return handleEvent(AIO_EVENT_DONE, nullptr); // hopefully a tail call |
| Lsuccess: |
| if (write_vc) { |
| CACHE_INCREMENT_DYN_STAT(cache_read_busy_success_stat); |
| } |
| SET_HANDLER(&CacheVC::openReadMain); |
| return callcont(CACHE_EVENT_OPEN_READ); |
| } |
| |
| // create the directory entry after the vector has been evacuated |
| // the volume lock has been taken when this function is called |
| int |
| CacheVC::openReadVecWrite(int /* event ATS_UNUSED */, Event * /* e ATS_UNUSED */) |
| { |
| cancel_trigger(); |
| set_io_not_in_progress(); |
| ink_assert(od); |
| od->writing_vec = false; |
| if (_action.cancelled) { |
| return openWriteCloseDir(EVENT_IMMEDIATE, nullptr); |
| } |
| { |
| CACHE_TRY_LOCK(lock, vol->mutex, mutex->thread_holding); |
| if (!lock.is_locked()) { |
| VC_SCHED_LOCK_RETRY(); |
| } |
| if (io.ok()) { |
| ink_assert(f.evac_vector); |
| ink_assert(frag_type == CACHE_FRAG_TYPE_HTTP); |
| ink_assert(!buf); |
| f.evac_vector = false; |
| last_collision = nullptr; |
| f.update = 0; |
| alternate_index = CACHE_ALT_INDEX_DEFAULT; |
| f.use_first_key = 0; |
| vio.op = VIO::READ; |
| dir_overwrite(&first_key, vol, &dir, &od->first_dir); |
| if (od->move_resident_alt) { |
| dir_insert(&od->single_doc_key, vol, &od->single_doc_dir); |
| } |
| int alt_ndx = HttpTransactCache::SelectFromAlternates(write_vector, &request, params); |
| vol->close_write(this); |
| if (alt_ndx >= 0) { |
| vector.clear(); |
| // we don't need to start all over again, since we already |
| // have the vector in memory. But this is simpler and this |
| // case is rare. |
| goto Lrestart; |
| } |
| } else { |
| vol->close_write(this); |
| } |
| } |
| |
| CACHE_INCREMENT_DYN_STAT(cache_read_failure_stat); |
| _action.continuation->handleEvent(CACHE_EVENT_OPEN_READ_FAILED, (void *)-ECACHE_ALT_MISS); |
| return free_CacheVC(this); |
| Lrestart: |
| SET_HANDLER(&CacheVC::openReadStartHead); |
| return openReadStartHead(EVENT_IMMEDIATE, nullptr); |
| } |
| |
| /* |
| This code follows CacheVC::openReadStartEarliest closely, |
| if you change this you might have to change that. |
| */ |
| int |
| CacheVC::openReadStartHead(int event, Event *e) |
| { |
| intptr_t err = ECACHE_NO_DOC; |
| Doc *doc = nullptr; |
| cancel_trigger(); |
| set_io_not_in_progress(); |
| if (_action.cancelled) { |
| return free_CacheVC(this); |
| } |
| { |
| CACHE_TRY_LOCK(lock, vol->mutex, mutex->thread_holding); |
| if (!lock.is_locked()) { |
| VC_SCHED_LOCK_RETRY(); |
| } |
| if (!buf) { |
| goto Lread; |
| } |
| if (!io.ok()) { |
| goto Ldone; |
| } |
| // an object needs to be outside the aggregation window in order to be |
| // be evacuated as it is read |
| if (!dir_agg_valid(vol, &dir)) { |
| // a directory entry which is no longer valid may have been overwritten |
| if (!dir_valid(vol, &dir)) { |
| last_collision = nullptr; |
| } |
| goto Lread; |
| } |
| doc = reinterpret_cast<Doc *>(buf->data()); |
| if (doc->magic != DOC_MAGIC) { |
| char tmpstring[CRYPTO_HEX_SIZE]; |
| if (is_action_tag_set("cache")) { |
| ink_release_assert(false); |
| } |
| if (doc->magic == DOC_CORRUPT) { |
| Warning("Head: Doc checksum does not match for %s", key.toHexStr(tmpstring)); |
| } else { |
| Warning("Head : Doc magic does not match for %s", key.toHexStr(tmpstring)); |
| } |
| // remove the dir entry |
| dir_delete(&key, vol, &dir); |
| // try going through the directory entries again |
| // in case the dir entry we deleted doesnt correspond |
| // to the key we are looking for. This is possible |
| // because of directory collisions |
| last_collision = nullptr; |
| goto Lread; |
| } |
| if (!(doc->first_key == key)) { |
| goto Lread; |
| } |
| if (f.lookup) { |
| goto Lookup; |
| } |
| earliest_dir = dir; |
| CacheHTTPInfo *alternate_tmp; |
| if (frag_type == CACHE_FRAG_TYPE_HTTP) { |
| uint32_t uml; |
| ink_assert(doc->hlen); |
| if (!doc->hlen) { |
| goto Ldone; |
| } |
| if ((uml = this->load_http_info(&vector, doc)) != doc->hlen) { |
| if (buf) { |
| HTTPCacheAlt *alt = reinterpret_cast<HTTPCacheAlt *>(doc->hdr()); |
| int32_t alt_length = 0; |
| // count should be reasonable, as vector is initialized and unlikely to be too corrupted |
| // by bad disk data - count should be the number of successfully unmarshalled alts. |
| for (int32_t i = 0; i < vector.count(); ++i) { |
| CacheHTTPInfo *info = vector.get(i); |
| if (info && info->m_alt) { |
| alt_length += info->m_alt->m_unmarshal_len; |
| } |
| } |
| Note("OpenReadHead failed for cachekey %X : vector inconsistency - " |
| "unmarshalled %d expecting %d in %d (base=%zu, ver=%d:%d) " |
| "- vector n=%d size=%d" |
| "first alt=%d[%s]", |
| key.slice32(0), uml, doc->hlen, doc->len, sizeof(Doc), doc->v_major, doc->v_minor, vector.count(), alt_length, |
| alt->m_magic, |
| (CACHE_ALT_MAGIC_ALIVE == alt->m_magic ? |
| "alive" : |
| CACHE_ALT_MAGIC_MARSHALED == alt->m_magic ? "serial" : CACHE_ALT_MAGIC_DEAD == alt->m_magic ? "dead" : "bogus")); |
| dir_delete(&key, vol, &dir); |
| } |
| err = ECACHE_BAD_META_DATA; |
| goto Ldone; |
| } |
| if (cache_config_select_alternate) { |
| alternate_index = HttpTransactCache::SelectFromAlternates(&vector, &request, params); |
| if (alternate_index < 0) { |
| err = ECACHE_ALT_MISS; |
| goto Ldone; |
| } |
| } else { |
| alternate_index = 0; |
| } |
| alternate_tmp = vector.get(alternate_index); |
| if (!alternate_tmp->valid()) { |
| if (buf) { |
| Note("OpenReadHead failed for cachekey %X : alternate inconsistency", key.slice32(0)); |
| dir_delete(&key, vol, &dir); |
| } |
| goto Ldone; |
| } |
| |
| alternate.copy_shallow(alternate_tmp); |
| alternate.object_key_get(&key); |
| doc_len = alternate.object_size_get(); |
| if (key == doc->key) { // is this my data? |
| f.single_fragment = doc->single_fragment(); |
| ink_assert(f.single_fragment); // otherwise need to read earliest |
| ink_assert(doc->hlen); |
| doc_pos = doc->prefix_len(); |
| next_CacheKey(&key, &doc->key); |
| } else { |
| f.single_fragment = false; |
| } |
| } else { |
| next_CacheKey(&key, &doc->key); |
| f.single_fragment = doc->single_fragment(); |
| doc_pos = doc->prefix_len(); |
| doc_len = doc->total_len; |
| } |
| |
| if (is_debug_tag_set("cache_read")) { // amc debug |
| char xt[CRYPTO_HEX_SIZE], yt[CRYPTO_HEX_SIZE]; |
| Debug("cache_read", "CacheReadStartHead - read %s target %s - %s %d of %" PRId64 " bytes, %d fragments", |
| doc->key.toHexStr(xt), key.toHexStr(yt), f.single_fragment ? "single" : "multi", doc->len, doc->total_len, |
| alternate.get_frag_offset_count()); |
| } |
| // the first fragment might have been gc'ed. Make sure the first |
| // fragment is there before returning CACHE_EVENT_OPEN_READ |
| if (!f.single_fragment) { |
| goto Learliest; |
| } |
| |
| if (vol->within_hit_evacuate_window(&dir) && |
| (!cache_config_hit_evacuate_size_limit || doc_len <= static_cast<uint64_t>(cache_config_hit_evacuate_size_limit))) { |
| DDebug("cache_hit_evac", "dir: %" PRId64 ", write: %" PRId64 ", phase: %d", dir_offset(&dir), |
| vol->offset_to_vol_offset(vol->header->write_pos), vol->header->phase); |
| f.hit_evacuate = 1; |
| } |
| |
| first_buf = buf; |
| vol->begin_read(this); |
| |
| goto Lsuccess; |
| |
| Lread: |
| // check for collision |
| // INKqa07684 - Cache::lookup returns CACHE_EVENT_OPEN_READ_FAILED. |
| // don't want to go through this BS of reading from a writer if |
| // its a lookup. In this case lookup will fail while the document is |
| // being written to the cache. |
| OpenDirEntry *cod = vol->open_read(&key); |
| if (cod && !f.read_from_writer_called) { |
| if (f.lookup) { |
| err = ECACHE_DOC_BUSY; |
| goto Ldone; |
| } |
| od = cod; |
| MUTEX_RELEASE(lock); |
| SET_HANDLER(&CacheVC::openReadFromWriter); |
| return handleEvent(EVENT_IMMEDIATE, nullptr); |
| } |
| if (dir_probe(&key, vol, &dir, &last_collision)) { |
| first_dir = dir; |
| int ret = do_read_call(&key); |
| if (ret == EVENT_RETURN) { |
| goto Lcallreturn; |
| } |
| return ret; |
| } |
| } |
| Ldone: |
| if (!f.lookup) { |
| CACHE_INCREMENT_DYN_STAT(cache_read_failure_stat); |
| _action.continuation->handleEvent(CACHE_EVENT_OPEN_READ_FAILED, (void *)-err); |
| } else { |
| CACHE_INCREMENT_DYN_STAT(cache_lookup_failure_stat); |
| _action.continuation->handleEvent(CACHE_EVENT_LOOKUP_FAILED, (void *)-err); |
| } |
| return free_CacheVC(this); |
| Lcallreturn: |
| return handleEvent(AIO_EVENT_DONE, nullptr); // hopefully a tail call |
| Lsuccess: |
| SET_HANDLER(&CacheVC::openReadMain); |
| return callcont(CACHE_EVENT_OPEN_READ); |
| Lookup: |
| CACHE_INCREMENT_DYN_STAT(cache_lookup_success_stat); |
| _action.continuation->handleEvent(CACHE_EVENT_LOOKUP, nullptr); |
| return free_CacheVC(this); |
| Learliest: |
| first_buf = buf; |
| buf = nullptr; |
| earliest_key = key; |
| last_collision = nullptr; |
| SET_HANDLER(&CacheVC::openReadStartEarliest); |
| return openReadStartEarliest(event, e); |
| } |
| |
| /* |
| Handle a directory delete event in case of some detected corruption. |
| */ |
| int |
| CacheVC::openReadDirDelete(int event, Event *e) |
| { |
| MUTEX_TRY_LOCK(lock, vol->mutex, mutex->thread_holding); |
| if (!lock.is_locked()) { |
| VC_SCHED_LOCK_RETRY(); |
| } |
| |
| dir_delete(&earliest_key, vol, &earliest_dir); |
| return calluser(VC_EVENT_ERROR); |
| } |
