pipeline.cc

Go to the documentation of this file.

#include "pipeline.h"

#include <algorithm>
#include <cstdlib>

#include "ingestion/task_chunk.h"
#include "ingestion/task_compress.h"
#include "ingestion/task_hash.h"
#include "ingestion/task_read.h"
#include "ingestion/task_register.h"
#include "ingestion/task_write.h"
#include "upload_facility.h"
#include "upload_spooler_definition.h"
#include "util/concurrency.h"
#include "util/exception.h"
#include "util/platform.h"
#include "util/string.h"

const uint64_t IngestionPipeline::kMaxPipelineMem = 1024 * 1024 * 1024;

IngestionPipeline::IngestionPipeline(
    upload::AbstractUploader *uploader,
    const upload::SpoolerDefinition &spooler_definition)
    : compression_algorithm_(spooler_definition.compression_alg)
    , hash_algorithm_(spooler_definition.hash_algorithm)
    , generate_legacy_bulk_chunks_(
          spooler_definition.generate_legacy_bulk_chunks)
    , chunking_enabled_(spooler_definition.use_file_chunking)
    , minimal_chunk_size_(spooler_definition.min_file_chunk_size)
    , average_chunk_size_(spooler_definition.avg_file_chunk_size)
    , maximal_chunk_size_(spooler_definition.max_file_chunk_size)
    , spawned_(false)
    , uploader_(uploader)
    , tube_ctr_inflight_pre_(kMaxFilesInFlight) {
  unsigned nfork_base = std::max(1U, GetNumberOfCpuCores() / 8);

  for (unsigned i = 0; i < nfork_base * kNforkRegister; ++i) {
    Tube<FileItem> *tube = new Tube<FileItem>();
    tubes_register_.TakeTube(tube);
    TaskRegister *task = new TaskRegister(tube, &tube_ctr_inflight_pre_,
                                          &tube_ctr_inflight_post_);
    task->RegisterListener(&IngestionPipeline::OnFileProcessed, this);
    tasks_register_.TakeConsumer(task);
  }
  tubes_register_.Activate();

  for (unsigned i = 0; i < nfork_base * kNforkWrite; ++i) {
    Tube<BlockItem> *t = new Tube<BlockItem>();
    tubes_write_.TakeTube(t);
    tasks_write_.TakeConsumer(new TaskWrite(t, &tubes_register_, uploader_));
  }
  tubes_write_.Activate();

  for (unsigned i = 0; i < nfork_base * kNforkHash; ++i) {
    Tube<BlockItem> *t = new Tube<BlockItem>();
    tubes_hash_.TakeTube(t);
    tasks_hash_.TakeConsumer(new TaskHash(t, &tubes_write_));
  }
  tubes_hash_.Activate();

  for (unsigned i = 0; i < nfork_base * kNforkCompress; ++i) {
    Tube<BlockItem> *t = new Tube<BlockItem>();
    tubes_compress_.TakeTube(t);
    tasks_compress_.TakeConsumer(
        new TaskCompress(t, &tubes_hash_, &item_allocator_));
  }
  tubes_compress_.Activate();

  for (unsigned i = 0; i < nfork_base * kNforkChunk; ++i) {
    Tube<BlockItem> *t = new Tube<BlockItem>();
    tubes_chunk_.TakeTube(t);
    tasks_chunk_.TakeConsumer(
        new TaskChunk(t, &tubes_compress_, &item_allocator_));
  }
  tubes_chunk_.Activate();

  uint64_t high = kMaxPipelineMem;
  high = std::min(high, platform_memsize() / 5);
  char *fixed_limit_mb = getenv("_CVMFS_SERVER_PIPELINE_MB");
  if (fixed_limit_mb != NULL) {
    high = String2Uint64(fixed_limit_mb) * 1024 * 1024;
  }
  uint64_t low = (high * 2) / 3;
  LogCvmfs(kLogCvmfs, kLogDebug,
           "pipeline memory thresholds %" PRIu64 "/%" PRIu64 " M",
           low / (1024 * 1024), high / (1024 * 1024));
  for (unsigned i = 0; i < nfork_base * kNforkRead; ++i) {
    TaskRead *task_read = new TaskRead(&tube_input_, &tubes_chunk_,
                                       &item_allocator_);
    task_read->SetWatermarks(low, high);
    tasks_read_.TakeConsumer(task_read);
  }
}


IngestionPipeline::~IngestionPipeline() {
  if (spawned_) {
    tasks_read_.Terminate();
    tasks_chunk_.Terminate();
    tasks_compress_.Terminate();
    tasks_hash_.Terminate();
    tasks_write_.Terminate();
    tasks_register_.Terminate();
  }
}


void IngestionPipeline::OnFileProcessed(
    const upload::SpoolerResult &spooler_result) {
  NotifyListeners(spooler_result);
}


void IngestionPipeline::Process(IngestionSource *source,
                                bool allow_chunking,
                                shash::Suffix hash_suffix) {
  FileItem *file_item = new FileItem(source,
                                     minimal_chunk_size_,
                                     average_chunk_size_,
                                     maximal_chunk_size_,
                                     compression_algorithm_,
                                     hash_algorithm_,
                                     hash_suffix,
                                     allow_chunking && chunking_enabled_,
                                     generate_legacy_bulk_chunks_);
  tube_ctr_inflight_post_.EnqueueBack(file_item);
  tube_ctr_inflight_pre_.EnqueueBack(file_item);
  tube_input_.EnqueueBack(file_item);
}


void IngestionPipeline::Spawn() {
  tasks_register_.Spawn();
  tasks_write_.Spawn();
  tasks_hash_.Spawn();
  tasks_compress_.Spawn();
  tasks_chunk_.Spawn();
  tasks_read_.Spawn();
  spawned_ = true;
}


void IngestionPipeline::WaitFor() { tube_ctr_inflight_post_.Wait(); }


//------------------------------------------------------------------------------


void TaskScrubbingCallback::Process(BlockItem *block_item) {
  FileItem *file_item = block_item->file_item();
  assert(file_item != NULL);
  assert(!file_item->path().empty());
  ChunkItem *chunk_item = block_item->chunk_item();
  assert(chunk_item != NULL);
  assert(chunk_item->is_bulk_chunk());

  switch (block_item->type()) {
    case BlockItem::kBlockData:
      delete block_item;
      break;

    case BlockItem::kBlockStop:
      assert(!chunk_item->hash_ptr()->IsNull());
      NotifyListeners(
          ScrubbingResult(file_item->path(), *chunk_item->hash_ptr()));
      delete block_item;
      delete chunk_item;
      delete file_item;
      tube_counter_->PopFront();
      break;

    default:
      PANIC(NULL);
  }
}


//------------------------------------------------------------------------------


ScrubbingPipeline::ScrubbingPipeline()
    : spawned_(false), tube_counter_(kMaxFilesInFlight) {
  unsigned nfork_base = std::max(1U, GetNumberOfCpuCores() / 8);

  for (unsigned i = 0; i < nfork_base * kNforkScrubbingCallback; ++i) {
    Tube<BlockItem> *tube = new Tube<BlockItem>();
    tubes_scrubbing_callback_.TakeTube(tube);
    TaskScrubbingCallback *task = new TaskScrubbingCallback(tube,
                                                            &tube_counter_);
    task->RegisterListener(&ScrubbingPipeline::OnFileProcessed, this);
    tasks_scrubbing_callback_.TakeConsumer(task);
  }
  tubes_scrubbing_callback_.Activate();

  for (unsigned i = 0; i < nfork_base * kNforkHash; ++i) {
    Tube<BlockItem> *t = new Tube<BlockItem>();
    tubes_hash_.TakeTube(t);
    tasks_hash_.TakeConsumer(new TaskHash(t, &tubes_scrubbing_callback_));
  }
  tubes_hash_.Activate();

  for (unsigned i = 0; i < nfork_base * kNforkChunk; ++i) {
    Tube<BlockItem> *t = new Tube<BlockItem>();
    tubes_chunk_.TakeTube(t);
    tasks_chunk_.TakeConsumer(new TaskChunk(t, &tubes_hash_, &item_allocator_));
  }
  tubes_chunk_.Activate();

  for (unsigned i = 0; i < nfork_base * kNforkRead; ++i) {
    TaskRead *task_read = new TaskRead(&tube_input_, &tubes_chunk_,
                                       &item_allocator_);
    task_read->SetWatermarks(kMemLowWatermark, kMemHighWatermark);
    tasks_read_.TakeConsumer(task_read);
  }
}


ScrubbingPipeline::~ScrubbingPipeline() {
  if (spawned_) {
    tasks_read_.Terminate();
    tasks_chunk_.Terminate();
    tasks_hash_.Terminate();
    tasks_scrubbing_callback_.Terminate();
  }
}


void ScrubbingPipeline::OnFileProcessed(
    const ScrubbingResult &scrubbing_result) {
  NotifyListeners(scrubbing_result);
}


void ScrubbingPipeline::Process(IngestionSource *source,
                                shash::Algorithms hash_algorithm,
                                shash::Suffix hash_suffix) {
  FileItem *file_item = new FileItem(source, 0, 0, 0, zlib::kNoCompression,
                                     hash_algorithm, hash_suffix,
                                     false, /* may_have_chunks */
                                     true /* hash_legacy_bulk_chunk */);
  tube_counter_.EnqueueBack(file_item);
  tube_input_.EnqueueBack(file_item);
}


void ScrubbingPipeline::Spawn() {
  tasks_scrubbing_callback_.Spawn();
  tasks_hash_.Spawn();
  tasks_chunk_.Spawn();
  tasks_read_.Spawn();
  spawned_ = true;
}


void ScrubbingPipeline::WaitFor() { tube_counter_.Wait(); }