remote.cc

   1 /* LBS: An LFS-inspired filesystem backup system
   2  * Copyright (C) 2008  Michael Vrable
   3  *
   4  * Backup data (segments and backup descriptors) may be stored on a remote
   5  * fileserver instead of locally.  The only local storage needed is for the
   6  * local database and some temporary space for staging files before they are
   7  * transferred to the remote server.
   8  *
   9  * Like encryption, remote storage is handled through the use of external
  10  * scripts that are called when a file is to be transferred. */
  11
  12 #include <assert.h>
  13 #include <fcntl.h>
  14 #include <unistd.h>
  15 #include <sys/types.h>
  16 #include <sys/stat.h>
  17 #include <sys/wait.h>
  18
  19 #include <list>
  20 #include <string>
  21
  22 #include "remote.h"
  23 #include "store.h"
  24
  25 using std::string;
  26
  27 RemoteStore::RemoteStore(const string &stagedir)
  28 {
  29     staging_dir = stagedir;
  30
  31     /* A background thread is created for each RemoteStore to manage the actual
  32      * transfers to a remote server.  The main program thread can enqueue
  33      * RemoteFile objects to be transferred asynchronously. */
  34     pthread_mutex_init(&lock, NULL);
  35     pthread_cond_init(&cond, NULL);
  36     terminate = false;
  37     busy = true;
  38     files_outstanding = 0;
  39
  40     if (pthread_create(&thread, NULL, RemoteStore::start_transfer_thread,
  41                        (void *)this) != 0) {
  42         fprintf(stderr, "Cannot create remote storage thread: %m\n");
  43         throw IOException("pthread_create");
  44     }
  45 }
  46
  47 /* The RemoteStore destructor will terminate the background transfer thread.
  48  * It will wait for all work to finish. */
  49 RemoteStore::~RemoteStore()
  50 {
  51     pthread_mutex_lock(&lock);
  52     terminate = true;
  53     pthread_cond_broadcast(&cond);
  54     pthread_mutex_unlock(&lock);
  55
  56     if (pthread_join(thread, NULL) != 0) {
  57         fprintf(stderr, "Warning: Unable to join storage thread: %m\n");
  58     }
  59
  60     assert(files_outstanding == 0);
  61
  62     pthread_cond_destroy(&cond);
  63     pthread_mutex_destroy(&lock);
  64 }
  65
  66 /* Prepare to write out a new file.  Returns a RemoteFile object.  The file
  67  * will initially be created in a temporary directory.  When the file is
  68  * written out, the RemoteFile object should be passed to RemoteStore::enqueue,
  69  * which will upload it to the remote server. */
  70 RemoteFile *RemoteStore::alloc_file(const string &name, const string &type)
  71 {
  72     fprintf(stderr, "Allocate file: %s\n", name.c_str());
  73     pthread_mutex_lock(&lock);
  74     files_outstanding++;
  75     pthread_mutex_unlock(&lock);
  76     return new RemoteFile(this, name, type, staging_dir + "/" + name);
  77 }
  78
  79 /* Request that a file be transferred to the remote server.  The actual
  80  * transfer will happen asynchronously in another thread.  The call to enqueue
  81  * may block, however, if there is a backlog of data to be transferred.
  82  * Ownership of the RemoteFile object is transferred; the RemoteStore will be
  83  * responsible for its destruction. */
  84 void RemoteStore::enqueue(RemoteFile *file)
  85 {
  86     fprintf(stderr, "Enqueue: %s\n", file->remote_path.c_str());
  87
  88     pthread_mutex_lock(&lock);
  89
  90     while (transfer_queue.size() >= MAX_QUEUE_SIZE)
  91         pthread_cond_wait(&cond, &lock);
  92
  93     transfer_queue.push_back(file);
  94     files_outstanding--;
  95     busy = true;
  96
  97     pthread_cond_broadcast(&cond);
  98     pthread_mutex_unlock(&lock);
  99 }
 100
 101 /* Wait for all transfers to finish. */
 102 void RemoteStore::sync()
 103 {
 104     fprintf(stderr, "RemoteStore::sync() start\n");
 105     pthread_mutex_lock(&lock);
 106
 107     while (busy)
 108         pthread_cond_wait(&cond, &lock);
 109
 110     pthread_mutex_unlock(&lock);
 111     fprintf(stderr, "RemoteStore::sync() end\n");
 112 }
 113
 114 void *RemoteStore::start_transfer_thread(void *arg)
 115 {
 116     RemoteStore *store = static_cast<RemoteStore *>(arg);
 117     store->transfer_thread();
 118     return NULL;
 119 }
 120
 121 /* Background thread for transferring backups to a remote server. */
 122 void RemoteStore::transfer_thread()
 123 {
 124     while (true) {
 125         RemoteFile *file = NULL;
 126
 127         // Wait for a file to transfer
 128         pthread_mutex_lock(&lock);
 129         while (transfer_queue.empty() && !terminate) {
 130             busy = false;
 131             pthread_cond_broadcast(&cond);
 132             pthread_cond_wait(&cond, &lock);
 133         }
 134         if (terminate && transfer_queue.empty()) {
 135             busy = false;
 136             pthread_cond_broadcast(&cond);
 137             pthread_mutex_unlock(&lock);
 138             break;
 139         }
 140         busy = true;
 141         file = transfer_queue.front();
 142         transfer_queue.pop_front();
 143         pthread_cond_broadcast(&cond);
 144         pthread_mutex_unlock(&lock);
 145
 146         // Transfer the file
 147         fprintf(stderr, "Start transfer: %s\n", file->remote_path.c_str());
 148         if (backup_script != "") {
 149             pid_t pid = fork();
 150             if (pid < 0) {
 151                 fprintf(stderr, "Unable to fork for upload script: %m\n");
 152                 throw IOException("fork: upload script");
 153             }
 154             if (pid == 0) {
 155                 string cmd = backup_script;
 156                 cmd += " " + file->local_path + " " + file->type + " "
 157                         + file->remote_path;
 158                 execlp("/bin/sh", "/bin/sh", "-c", cmd.c_str(), NULL);
 159                 throw IOException("exec failed");
 160             }
 161
 162             int status = 0;
 163             waitpid(pid, &status, 0);
 164             if (!WIFEXITED(status) || WEXITSTATUS(status) != 0) {
 165                 fprintf(stderr, "Warning: error code from upload script: %d\n",
 166                         status);
 167             }
 168
 169             if (unlink(file->local_path.c_str()) < 0) {
 170                 fprintf(stderr, "Warning: Deleting temporary file %s: %m\n",
 171                         file->local_path.c_str());
 172             }
 173         }
 174         fprintf(stderr, "Finish transfer: %s\n", file->remote_path.c_str());
 175
 176         delete file;
 177     }
 178 }
 179
 180 RemoteFile::RemoteFile(RemoteStore *remote,
 181                        const string &name, const string &type,
 182                        const string &local_path)
 183 {
 184     remote_store = remote;
 185     this->type = type;
 186     this->local_path = local_path;
 187     this->remote_path = name;
 188
 189     fd = open(local_path.c_str(), O_WRONLY | O_CREAT, 0666);
 190     if (fd < 0)
 191         throw IOException("Error opening output file");
 192 }