upload_facility.h

Go to the documentation of this file.

#ifndef CVMFS_UPLOAD_FACILITY_H_
#define CVMFS_UPLOAD_FACILITY_H_

#include <fcntl.h>
#include <stdint.h>

#include <string>

#include "ingestion/ingestion_source.h"
#include "ingestion/task.h"
#include "repository_tag.h"
#include "statistics.h"
#include "upload_spooler_definition.h"
#include "util/atomic.h"
#include "util/concurrency.h"
#include "util/posix.h"
#include "util/tube.h"

namespace upload {

struct UploadCounters {
  perf::Counter *n_chunks_added;
  perf::Counter *n_chunks_duplicated;
  perf::Counter *n_catalogs_added;
  perf::Counter *sz_uploaded_bytes;
  perf::Counter *sz_uploaded_catalog_bytes;

  explicit UploadCounters(perf::StatisticsTemplate statistics) {
    n_chunks_added = statistics.RegisterOrLookupTemplated(
        "n_chunks_added", "Number of new chunks added");
    n_chunks_duplicated = statistics.RegisterOrLookupTemplated(
        "n_chunks_duplicated", "Number of duplicated chunks added");
    n_catalogs_added = statistics.RegisterOrLookupTemplated(
        "n_catalogs_added", "Number of new catalogs added");
    sz_uploaded_bytes = statistics.RegisterOrLookupTemplated(
        "sz_uploaded_bytes", "Number of uploaded bytes");
    sz_uploaded_catalog_bytes = statistics.RegisterOrLookupTemplated(
        "sz_uploaded_catalog_bytes", "Number of uploaded bytes for catalogs");
  }
};  // UploadCounters

struct UploaderResults {
  enum Type {
    kFileUpload,
    kBufferUpload,
    kChunkCommit,
    kRemove,
    kLookup
  };

  UploaderResults(const int return_code, const std::string &local_path)
      : type(kFileUpload), return_code(return_code), local_path(local_path) { }

  explicit UploaderResults(Type t, const int return_code)
      : type(t), return_code(return_code), local_path("") { }

  UploaderResults() : type(kRemove), return_code(0) { }

  const Type type;
  const int return_code;
  const std::string local_path;
};

struct UploadStreamHandle;

class AbstractUploader
    : public PolymorphicConstruction<AbstractUploader, SpoolerDefinition>,
      public Callbackable<UploaderResults>,
      public SingleCopy {
  friend class TaskUpload;

 public:
  struct UploadBuffer {
    UploadBuffer() : size(0), data(NULL) { }
    UploadBuffer(uint64_t s, const void *d) : size(s), data(d) { }
    uint64_t size;
    const void *data;
  };

  struct JobStatus {
    enum State {
      kOk,
      kTerminate,
      kNoJobs
    };
  };

  struct UploadJob {
    enum Type {
      Upload,
      Commit,
      Terminate
    };

    UploadJob(UploadStreamHandle *handle, UploadBuffer buffer,
              const CallbackTN *callback = NULL);
    UploadJob(UploadStreamHandle *handle, const shash::Any &content_hash);

    UploadJob()
        : type(Terminate)
        , stream_handle(NULL)
        , tag_(0)
        , buffer()
        , callback(NULL) { }

    static UploadJob *CreateQuitBeacon() { return new UploadJob(); }
    bool IsQuitBeacon() { return type == Terminate; }

    Type type;
    UploadStreamHandle *stream_handle;
    int64_t tag_;
    int64_t tag() { return tag_; }

    // type==Upload specific fields
    UploadBuffer buffer;
    const CallbackTN *callback;

    // type==Commit specific fields
    shash::Any content_hash;
  };

  virtual ~AbstractUploader() { assert(!tasks_upload_.is_active()); }

  virtual std::string name() const = 0;

  virtual bool Create() = 0;

  virtual unsigned GetNumTasks() const { return num_upload_tasks_; }

  virtual bool Initialize();

  virtual bool FinalizeSession(bool commit, const std::string &old_root_hash,
                               const std::string &new_root_hash,
                               const RepositoryTag &tag);

  void TearDown();

  void UploadFile(const std::string &local_path,
                  const std::string &remote_path,
                  const CallbackTN *callback = NULL) {
    ++jobs_in_flight_;
    FileIngestionSource source(local_path);
    DoUpload(remote_path, &source, callback);
  }

  void UploadIngestionSource(const std::string &remote_path,
                             IngestionSource *source,
                             const CallbackTN *callback = NULL) {
    ++jobs_in_flight_;
    DoUpload(remote_path, source, callback);
  }

  virtual UploadStreamHandle *InitStreamedUpload(
      const CallbackTN *callback) = 0;

  void ScheduleUpload(UploadStreamHandle *handle,
                      UploadBuffer buffer,
                      const CallbackTN *callback = NULL) {
    ++jobs_in_flight_;
    tubes_upload_.Dispatch(new UploadJob(handle, buffer, callback));
  }

  void ScheduleCommit(UploadStreamHandle *handle,
                      const shash::Any &content_hash) {
    ++jobs_in_flight_;
    tubes_upload_.Dispatch(new UploadJob(handle, content_hash));
  }

  void RemoveAsync(const std::string &file_to_delete) {
    ++jobs_in_flight_;
    DoRemoveAsync(file_to_delete);
  }

  void RemoveAsync(const shash::Any &hash_to_delete) {
    RemoveAsync("data/" + hash_to_delete.MakePath());
  }

  int64_t GetObjectSize(const shash::Any &hash) {
    return DoGetObjectSize("data/" + hash.MakePath());
  }

  virtual bool Peek(const std::string &path) = 0;

  virtual bool Mkdir(const std::string &path) = 0;

  virtual bool PlaceBootstrappingShortcut(const shash::Any &object) = 0;

  virtual void WaitForUpload() const;

  virtual unsigned int GetNumberOfErrors() const = 0;
  static void RegisterPlugins();
  void InitCounters(perf::StatisticsTemplate *statistics);

 protected:
  typedef Callbackable<UploaderResults>::CallbackTN *CallbackPtr;

  explicit AbstractUploader(const SpoolerDefinition &spooler_definition);

  virtual void DoUpload(const std::string &remote_path,
                        IngestionSource *source,
                        const CallbackTN *callback) = 0;

  virtual void StreamedUpload(UploadStreamHandle *handle,
                              UploadBuffer buffer,
                              const CallbackTN *callback) = 0;

  virtual void FinalizeStreamedUpload(UploadStreamHandle *handle,
                                      const shash::Any &content_hash) = 0;


  virtual void DoRemoveAsync(const std::string &file_to_delete) = 0;

  virtual int64_t DoGetObjectSize(const std::string &file_name) = 0;

  void Respond(const CallbackTN *callback,
               const UploaderResults &result) const {
    if (callback != NULL) {
      (*callback)(result);
      delete callback;
    }

    --jobs_in_flight_;
  }

  int CreateAndOpenTemporaryChunkFile(std::string *path) const;

  const SpoolerDefinition &spooler_definition() const {
    return spooler_definition_;
  }

  void CountUploadedChunks() const;
  void DecUploadedChunks() const;
  void CountUploadedBytes(int64_t bytes_written) const;
  void CountDuplicates() const;
  void CountUploadedCatalogs() const;
  void CountUploadedCatalogBytes(int64_t bytes_written) const;

 protected:
  void IncJobsInFlight() { ++jobs_in_flight_; }

 private:
  const SpoolerDefinition spooler_definition_;

  unsigned num_upload_tasks_;
  mutable SynchronizingCounter<int32_t> jobs_in_flight_;
  TubeGroup<UploadJob> tubes_upload_;
  TubeConsumerGroup<UploadJob> tasks_upload_;
  mutable UniquePtr<UploadCounters> counters_;
};  // class AbstractUploader


class TaskUpload : public TubeConsumer<AbstractUploader::UploadJob> {
 public:
  explicit TaskUpload(AbstractUploader *uploader,
                      Tube<AbstractUploader::UploadJob> *tube)
      : TubeConsumer<AbstractUploader::UploadJob>(tube), uploader_(uploader) { }

 protected:
  virtual void Process(AbstractUploader::UploadJob *upload_job);

 private:
  AbstractUploader *uploader_;
};


struct UploadStreamHandle {
  typedef AbstractUploader::CallbackTN CallbackTN;
  static atomic_int64 g_upload_stream_tag;

  explicit UploadStreamHandle(const CallbackTN *commit_callback)
      : commit_callback(commit_callback)
      , tag(atomic_xadd64(&g_upload_stream_tag, 1)) { }
  virtual ~UploadStreamHandle() { }

  const CallbackTN *commit_callback;

  int64_t tag;

  std::string remote_path;  // override remote location of the object
};

}  // namespace upload

#endif  // CVMFS_UPLOAD_FACILITY_H_