CacheVol.cc

Go to the documentation of this file.

/** @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"

#define SCAN_BUF_SIZE      RECOVERY_SIZE
#define SCAN_WRITER_LOCK_MAX_RETRY 5

Action *
Cache::scan(Continuation * cont, char *hostname, int host_len, int KB_per_second)
{
  Debug("cache_scan_truss", "inside scan");
  if (!CacheProcessor::IsCacheReady(CACHE_FRAG_TYPE_HTTP)) {
    cont->handleEvent(CACHE_EVENT_SCAN_FAILED, 0);
    return ACTION_RESULT_DONE;
  }

  CacheVC *c = new_CacheVC(cont);
  c->vol = NULL;
  /* do we need to make a copy */
  c->hostname = hostname;
  c->host_len = host_len;
  c->base_stat = cache_scan_active_stat;
  c->buf = new_IOBufferData(BUFFER_SIZE_FOR_XMALLOC(SCAN_BUF_SIZE), MEMALIGNED);
  c->scan_msec_delay = (SCAN_BUF_SIZE / KB_per_second);
  c->offset = 0;
  SET_CONTINUATION_HANDLER(c, &CacheVC::scanVol);
  eventProcessor.schedule_in(c, HRTIME_MSECONDS(c->scan_msec_delay));
  cont->handleEvent(CACHE_EVENT_SCAN, c);
  return &c->_action;
}

int
CacheVC::scanVol(int /* event ATS_UNUSED */, Event * /* e ATS_UNUSED */)
{
  Debug("cache_scan_truss", "inside %p:scanVol", this);
  if (_action.cancelled)
    return free_CacheVC(this);
  CacheHostRecord *rec = &theCache->hosttable->gen_host_rec;
  if (host_len) {
    CacheHostResult res;
    theCache->hosttable->Match(hostname, host_len, &res);
    if (res.record)
      rec = res.record;
  }
  if (!vol) {
    if (!rec->num_vols)
      goto Ldone;
    vol = rec->vols[0];
  } else {
    for (int i = 0; i < rec->num_vols - 1; i++)
      if (vol == rec->vols[i]) {
        vol = rec->vols[i + 1];
        goto Lcont;
      }
    goto Ldone;
  }
Lcont:
  fragment = 0;
  SET_HANDLER(&CacheVC::scanObject);
  eventProcessor.schedule_in(this, HRTIME_MSECONDS(scan_msec_delay));
  return EVENT_CONT;
Ldone:
  _action.continuation->handleEvent(CACHE_EVENT_SCAN_DONE, NULL);
  return free_CacheVC(this);
}

/* Next block with some data in it in this partition.  Returns end of partition if no more
 * locations.
 *
 * d - Vol
 * vol_map - precalculated map
 * offset - offset to start looking at (and data at this location has not been read yet). */
static off_t next_in_map(Vol *d, char *vol_map, off_t offset)
{
  off_t start_offset = vol_offset_to_offset(d, 0);
  off_t new_off = (offset - start_offset);
  off_t vol_len = vol_relative_length(d, start_offset);

  while (new_off < vol_len && !vol_map[new_off / SCAN_BUF_SIZE]) new_off += SCAN_BUF_SIZE;
  if (new_off >= vol_len) return vol_len + start_offset;
  return new_off + start_offset;
}

// Function in CacheDir.cc that we need for make_vol_map().
int
dir_bucket_loop_fix(Dir *start_dir, int s, Vol *d);

// TODO: If we used a bit vector, we could make a smaller map structure.
// TODO: If we saved a high water mark we could have a smaller buf, and avoid searching it
// when we are asked about the highest interesting offset.
/* Make map of what blocks in partition are used.
 *
 * d - Vol to make a map of. */
static char *make_vol_map(Vol *d)
{
  // Map will be one byte for each SCAN_BUF_SIZE bytes.
  off_t start_offset = vol_offset_to_offset(d, 0);
  off_t vol_len = vol_relative_length(d, start_offset);
  size_t map_len = (vol_len + (SCAN_BUF_SIZE - 1)) / SCAN_BUF_SIZE;
  char *vol_map = (char *)ats_malloc(map_len);

  memset(vol_map, 0, map_len);

  // Scan directories.
  // Copied from dir_entries_used() and modified to fill in the map instead.
  for (int s = 0; s < d->segments; s++) {
    Dir *seg = dir_segment(s, d);
    for (int b = 0; b < d->buckets; b++) {
      Dir *e = dir_bucket(b, seg);
      if (dir_bucket_loop_fix(e, s, d)) {
        break;
      }
      while (e) {
        if (dir_offset(e)) {
            off_t offset = vol_offset(d, e) - start_offset;
            if (offset <= vol_len) vol_map[offset / SCAN_BUF_SIZE] = 1;
        }
        e = next_dir(e, seg);
        if (!e)
          break;
      }
    }
  }
  return vol_map;
}

int
CacheVC::scanObject(int /* event ATS_UNUSED */, Event * /* e ATS_UNUSED */)
{
  Debug("cache_scan_truss", "inside %p:scanObject", this);

  Doc *doc = NULL;
  void *result = NULL;
#ifdef HTTP_CACHE
  int hlen = 0;
  char hname[500];
  bool hostinfo_copied = false;
#endif
  off_t next_object_len = 0;
  bool might_need_overlap_read = false;

  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) {
    Debug("cache_scan_truss", "delay %p:scanObject", this);
    mutex->thread_holding->schedule_in_local(this, HRTIME_MSECONDS(cache_config_mutex_retry_delay));
    return EVENT_CONT;
  }

  if (!fragment) {               // initialize for first read
    fragment = 1;
    scan_vol_map = make_vol_map(vol);
    io.aiocb.aio_offset = next_in_map(vol, scan_vol_map, vol_offset_to_offset(vol, 0));
    if (io.aiocb.aio_offset >= (off_t)(vol->skip + vol->len))
      goto Ldone;
    io.aiocb.aio_nbytes = SCAN_BUF_SIZE;
    io.aiocb.aio_buf = buf->data();
    io.action = this;
    io.thread = AIO_CALLBACK_THREAD_ANY;
    Debug("cache_scan_truss", "read %p:scanObject", this);
    goto Lread;
  }

  if ((size_t)io.aio_result != (size_t) io.aiocb.aio_nbytes) {
    result = (void *) -ECACHE_READ_FAIL;
    goto Ldone;
  }

  doc = (Doc *) (buf->data() + offset);
  // If there is data in the buffer before the start that is from a partial object read previously
  // Fix things as if we read it this time.
  if (scan_fix_buffer_offset) {
    io.aio_result += scan_fix_buffer_offset;
    io.aiocb.aio_nbytes += scan_fix_buffer_offset;
    io.aiocb.aio_offset -= scan_fix_buffer_offset;
    io.aiocb.aio_buf = (char *)io.aiocb.aio_buf - scan_fix_buffer_offset;
    scan_fix_buffer_offset = 0;
  }
  while ((off_t)((char *) doc - buf->data()) + next_object_len < (off_t)io.aiocb.aio_nbytes) {
    might_need_overlap_read = false;
    doc = (Doc *) ((char *) doc + next_object_len);
    next_object_len = vol->round_to_approx_size(doc->len);
#ifdef HTTP_CACHE
    int i;
    bool changed;

    if (doc->magic != DOC_MAGIC) {
      next_object_len = CACHE_BLOCK_SIZE;
      Debug("cache_scan_truss", "blockskip %p:scanObject", this);
      continue;
    }
      
    if (doc->doc_type != CACHE_FRAG_TYPE_HTTP || !doc->hlen)
      goto Lskip;

    last_collision = NULL;
    while (1) {
      if (!dir_probe(&doc->first_key, vol, &dir, &last_collision))
        goto Lskip;
      if (!dir_agg_valid(vol, &dir) || !dir_head(&dir) ||
          (vol_offset(vol, &dir) != io.aiocb.aio_offset + ((char *) doc - buf->data())))
        continue;
      break;
    }
    if (doc->data() - buf->data() > (int) io.aiocb.aio_nbytes) {
      might_need_overlap_read = true;
      goto Lskip;
    }
    {
      char *tmp = doc->hdr();
      int len = doc->hlen;
      while (len > 0) {
        int r = HTTPInfo::unmarshal(tmp, len, buf._ptr());
        if (r < 0) {
          ink_assert(!"CacheVC::scanObject unmarshal failed");
          goto Lskip;
        }
        len -= r;
        tmp += r;
      }
    }
    if (this->load_http_info(&vector, doc) != doc->hlen)
      goto Lskip;
    changed = false;
    hostinfo_copied = 0;
    for (i = 0; i < vector.count(); i++) {
      if (!vector.get(i)->valid())
        goto Lskip;
      if (!hostinfo_copied) {
        memccpy(hname, vector.get(i)->request_get()->host_get(&hlen), 0, 500);
        hname[hlen] = 0;
        Debug("cache_scan", "hostname = '%s', hostlen = %d", hname, hlen);
        hostinfo_copied = 1;
      }
      vector.get(i)->object_key_get(&key);
      alternate_index = i;
      // verify that the earliest block exists, reducing 'false hit' callbacks
      if (!(key == doc->key)) {
        last_collision = NULL;
        if (!dir_probe(&key, vol, &earliest_dir, &last_collision))
          continue;
      }
      earliest_key = key;
      int result1 = _action.continuation->handleEvent(CACHE_EVENT_SCAN_OBJECT, vector.get(i));
      switch (result1) {
      case CACHE_SCAN_RESULT_CONTINUE:
        continue;
      case CACHE_SCAN_RESULT_DELETE:
        changed = true;
        vector.remove(i, true);
        i--;
        continue;
      case CACHE_SCAN_RESULT_DELETE_ALL_ALTERNATES:
        changed = true;
        vector.clear();
        i = 0;
        break;
      case CACHE_SCAN_RESULT_UPDATE:
        ink_assert(alternate_index >= 0);
        vector.insert(&alternate, alternate_index);
        if (!vector.get(alternate_index)->valid())
          continue;
        changed = true;
        continue;
      case EVENT_DONE:
        goto Lcancel;
      default:
        ink_assert(!"unexpected CACHE_SCAN_RESULT");
        continue;
      }
    }
    if (changed) {
      if (!vector.count()) {
        ink_assert(hostinfo_copied);
        SET_HANDLER(&CacheVC::scanRemoveDone);
        // force remove even if there is a writer
        cacheProcessor.remove(this, &doc->first_key, true, CACHE_FRAG_TYPE_HTTP, true, false, (char *) hname, hlen);
        return EVENT_CONT;
      } else {
        offset = (char *) doc - buf->data();
        write_len = 0;
        frag_type = CACHE_FRAG_TYPE_HTTP;
        f.use_first_key = 1;
        f.evac_vector = 1;
        first_key = key = doc->first_key;
        alternate_index = CACHE_ALT_REMOVED;
        earliest_key = zero_key;
        writer_lock_retry = 0;
        SET_HANDLER(&CacheVC::scanOpenWrite);
        return scanOpenWrite(EVENT_NONE, 0);
      }
    }
    continue;
  Lskip:;
#endif
  }
#ifdef HTTP_CACHE
  vector.clear();
#endif
    // If we had an object that went past the end of the buffer, and it is small enough to fix,
    // fix it.
  if (might_need_overlap_read &&
      ((off_t)((char *) doc - buf->data()) + next_object_len > (off_t)io.aiocb.aio_nbytes) &&
      next_object_len > 0) {
    off_t partial_object_len = io.aiocb.aio_nbytes - ((char *)doc - buf->data());
    // Copy partial object to beginning of the buffer.
    memmove(buf->data(), (char *)doc, partial_object_len);
    io.aiocb.aio_offset += io.aiocb.aio_nbytes;
    io.aiocb.aio_nbytes = SCAN_BUF_SIZE - partial_object_len;
    io.aiocb.aio_buf = buf->data() + partial_object_len;
    scan_fix_buffer_offset = partial_object_len;
  } else { // Normal case, where we ended on a object boundary.
    io.aiocb.aio_offset += ((char *)doc - buf->data()) + next_object_len;
    Debug("cache_scan_truss", "next %p:scanObject %" PRId64, this, (int64_t)io.aiocb.aio_offset);
    io.aiocb.aio_offset = next_in_map(vol, scan_vol_map, io.aiocb.aio_offset);
    Debug("cache_scan_truss", "next_in_map %p:scanObject %" PRId64, this, (int64_t)io.aiocb.aio_offset);
    io.aiocb.aio_nbytes = SCAN_BUF_SIZE;
    io.aiocb.aio_buf = buf->data();
    scan_fix_buffer_offset = 0;
  }

  if (io.aiocb.aio_offset >= vol->skip + vol->len) {
    SET_HANDLER(&CacheVC::scanVol);
    eventProcessor.schedule_in(this, HRTIME_MSECONDS(scan_msec_delay));
    return EVENT_CONT;
  }

Lread:
  io.aiocb.aio_fildes = vol->fd;
  if ((off_t)(io.aiocb.aio_offset + io.aiocb.aio_nbytes) > (off_t)(vol->skip + vol->len))
    io.aiocb.aio_nbytes = vol->skip + vol->len - io.aiocb.aio_offset;
  offset = 0;
  ink_assert(ink_aio_read(&io) >= 0);
  Debug("cache_scan_truss", "read %p:scanObject %" PRId64 " %zu", this,
        (int64_t)io.aiocb.aio_offset, (size_t)io.aiocb.aio_nbytes);
  return EVENT_CONT;

Ldone:
   Debug("cache_scan_truss", "done %p:scanObject", this);
  _action.continuation->handleEvent(CACHE_EVENT_SCAN_DONE, result);
#ifdef HTTP_CACHE
Lcancel:
#endif
  return free_CacheVC(this);
}

int
CacheVC::scanRemoveDone(int /* event ATS_UNUSED */, Event * /* e ATS_UNUSED */)
{
  Debug("cache_scan_truss", "inside %p:scanRemoveDone", this);
  Debug("cache_scan", "remove done.");
#ifdef HTTP_CACHE
  alternate.destroy();
#endif
  SET_HANDLER(&CacheVC::scanObject);
  return handleEvent(EVENT_IMMEDIATE, 0);
}

int
CacheVC::scanOpenWrite(int /* event ATS_UNUSED */, Event * /* e ATS_UNUSED */)
{
  Debug("cache_scan_truss", "inside %p:scanOpenWrite", this);
  cancel_trigger();
  // get volume lock
  if (writer_lock_retry > SCAN_WRITER_LOCK_MAX_RETRY) {
    int r = _action.continuation->handleEvent(CACHE_EVENT_SCAN_OPERATION_BLOCKED, 0);
    Debug("cache_scan", "still havent got the writer lock, asking user..");
    switch (r) {
    case CACHE_SCAN_RESULT_RETRY:
      writer_lock_retry = 0;
      break;
    case CACHE_SCAN_RESULT_CONTINUE:
      SET_HANDLER(&CacheVC::scanObject);
      return scanObject(EVENT_IMMEDIATE, 0);
    }
  }
  int ret = 0;
  {
    CACHE_TRY_LOCK(lock, vol->mutex, mutex->thread_holding);
    if (!lock) {
      Debug("cache_scan", "vol->mutex %p:scanOpenWrite", this);
      VC_SCHED_LOCK_RETRY();
    }

    Debug("cache_scan", "trying for writer lock");
    if (vol->open_write(this, false, 1)) {
      writer_lock_retry++;
      SET_HANDLER(&CacheVC::scanOpenWrite);
      mutex->thread_holding->schedule_in_local(this, scan_msec_delay);
      return EVENT_CONT;
    }

    ink_assert(this->od);
    // put all the alternates in the open directory vector
    int alt_count = vector.count();
    for (int i = 0; i < alt_count; i++) {
      write_vector->insert(vector.get(i));
    }
    od->writing_vec = 1;
    vector.clear(false);
    // check that the directory entry was not overwritten
    // if so return failure
    Debug("cache_scan", "got writer lock");
    Dir *l = NULL;
    Dir d;
    Doc *doc = (Doc *) (buf->data() + offset);
    offset = (char *) doc - buf->data() + vol->round_to_approx_size(doc->len);
    // if the doc contains some data, then we need to create
    // a new directory entry for this fragment. Remember the
    // offset and the key in earliest_key
    dir_assign(&od->first_dir, &dir);
    if (doc->total_len) {
      dir_assign(&od->single_doc_dir, &dir);
      dir_set_tag(&od->single_doc_dir, doc->key.slice32(2));
      od->single_doc_key = doc->key;
      od->move_resident_alt = 1;
    }

    while (1) {
      if (!dir_probe(&first_key, vol, &d, &l)) {
        vol->close_write(this);
        _action.continuation->handleEvent(CACHE_EVENT_SCAN_OPERATION_FAILED, 0);
        SET_HANDLER(&CacheVC::scanObject);
        return handleEvent(EVENT_IMMEDIATE, 0);
      }
      if (memcmp(&dir, &d, SIZEOF_DIR)) {
        Debug("cache_scan", "dir entry has changed");
        continue;
      }
      break;
    }

    // the document was not modified
    // we are safe from now on as we hold the
    // writer lock on the doc
    if (f.evac_vector)
      header_len = write_vector->marshal_length();
    SET_HANDLER(&CacheVC::scanUpdateDone);
    ret = do_write_call();
  }
  if (ret == EVENT_RETURN)
    return handleEvent(AIO_EVENT_DONE, 0);
  return ret;
}

int
CacheVC::scanUpdateDone(int /* event ATS_UNUSED */, Event * /* e ATS_UNUSED */)
{
  Debug("cache_scan_truss", "inside %p:scanUpdateDone", this);
  cancel_trigger();
  // get volume lock
  CACHE_TRY_LOCK(lock, vol->mutex, mutex->thread_holding);
  if (lock) {
    // insert a directory entry for the previous fragment
    dir_overwrite(&first_key, vol, &dir, &od->first_dir, false);
    if (od->move_resident_alt) {
      dir_insert(&od->single_doc_key, vol, &od->single_doc_dir);
    }
    ink_assert(vol->open_read(&first_key));
    ink_assert(this->od);
    vol->close_write(this);
    SET_HANDLER(&CacheVC::scanObject);
    return handleEvent(EVENT_IMMEDIATE, 0);
  } else {
    mutex->thread_holding->schedule_in_local(this, HRTIME_MSECONDS(cache_config_mutex_retry_delay));
    return EVENT_CONT;
  }
}