Initial revision

1 files changed, 264 insertions, 0 deletions

diff --git a/rsync/tube.c b/rsync/tube.c
new file mode 100644
index 0000000..0b82adc
--- a/dev/null
+++ b/rsync/tube.c
@@ -0,0 +1,264 @@
+/*= -*- c-basic-offset: 4; indent-tabs-mode: nil; -*-
+ *
+ * librsync -- dynamic caching and delta update in HTTP
+ * $Id$
+ * 
+ * Copyright (C) 2000, 2001 by Martin Pool <mbp@samba.org>
+ * 
+ * This program is free software; you can redistribute it and/or modify
+ * it under the terms of the GNU Lesser General Public License as published by
+ * the Free Software Foundation; either version 2.1 of the License, or
+ * (at your option) any later version.
+ * 
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU Lesser General Public License for more details.
+ * 
+ * You should have received a copy of the GNU Lesser General Public License
+ * along with this program; if not, write to the Free Software
+ * Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA.
+ */
+
+                              /*
+                               | Where a calculator on the ENIAC is
+                               | equpped with 18,000 vaccuum tubes and
+                               | weighs 30 tons, computers in the
+                               | future may have only 1,000 vaccuum
+                               | tubes and perhaps weigh 1 1/2
+                               | tons.
+                               |   -- Popular Mechanics, March 1949
+                               */
+
+
+/* tube: a somewhat elastic but fairly small buffer for data passing
+ * through a stream.
+ *
+ * In most cases the iter can adjust to send just as much data will
+ * fit.  In some cases that would be too complicated, because it has
+ * to transmit an integer or something similar.  So in that case we
+ * stick whatever won't fit into a small buffer.
+ *
+ * A tube can contain some literal data to go out (typically command
+ * bytes), and also an instruction to copy data from the stream's
+ * input or from some other location.  Both literal data and a copy
+ * command can be queued at the same time, but only in that order and
+ * at most one of each. */
+
+
+/*
+ * TODO: As an optimization, write it directly to the stream if
+ * possible.  But for simplicity don't do that yet.
+ *
+ * TODO: I think our current copy code will lock up if the application
+ * only ever calls us with either input or output buffers, and not
+ * both.  So I guess in that case we might need to copy into some
+ * temporary buffer space, and then back out again later.
+ */
+
+
+#include <config_rsync.h>
+
+#include <assert.h>
+#include <stdlib.h>
+#include <string.h>
+#include <stdio.h>
+
+#include "rsync.h"
+#include "trace.h"
+#include "util.h"
+#include "job.h"
+#include "stream.h"
+
+
+static void rs_tube_catchup_write(rs_job_t *job)
+{
+    rs_buffers_t *stream = job->stream;
+    int len, remain;
+
+    len = job->write_len;
+    assert(len > 0);
+
+    assert(len > 0);
+    if ((size_t) len > stream->avail_out)
+        len = stream->avail_out;
+
+    if (!stream->avail_out) {
+        rs_trace("no output space available");
+        return;
+    }
+
+    memcpy(stream->next_out, job->write_buf, len);
+    stream->next_out += len;
+    stream->avail_out -= len;
+
+    remain = job->write_len - len;
+    rs_trace("transmitted %d write bytes from tube, "
+             "%d remain to be sent",
+             len, remain);
+
+    if (remain > 0) {
+        /* Still something left in the tube... */
+        memmove(job->write_buf, job->write_buf + len, remain);
+    } else {
+        assert(remain == 0);
+    }
+
+    job->write_len = remain;
+}
+
+
+/**
+ * Execute a copy command, taking data from the scoop.
+ *
+ * \sa rs_tube_catchup_copy()
+ */
+static void
+rs_tube_copy_from_scoop(rs_job_t *job)
+{
+    size_t       this_len;
+    rs_buffers_t *stream = job->stream;
+
+    this_len = job->copy_len;
+    if (this_len > job->scoop_avail) {
+        this_len = job->scoop_avail;
+    }
+    if (this_len > stream->avail_out) {
+        this_len = stream->avail_out;
+    }
+
+    memcpy(stream->next_out, job->scoop_next, this_len);
+
+    stream->next_out += this_len;
+    stream->avail_out -= this_len;
+        
+    job->scoop_avail -= this_len;
+    job->scoop_next += this_len;
+
+    job->copy_len -= this_len;
+
+    rs_trace("caught up on %ld copied bytes from scoop, %ld remain there, "
+             "%ld remain to be copied", 
+             (long) this_len, (long) job->scoop_avail, (long) job->copy_len);
+}
+
+
+
+/**
+ * Catch up on an outstanding copy command.
+ *
+ * Takes data from the scoop, and the input (in that order), and
+ * writes as much as will fit to the output, up to the limit of the
+ * outstanding copy.
+ */
+static void rs_tube_catchup_copy(rs_job_t *job)
+{
+    rs_buffers_t *stream = job->stream;
+
+    assert(job->write_len == 0);
+    assert(job->copy_len > 0);
+
+    if (job->scoop_avail  && job->copy_len) {
+        /* there's still some data in the scoop, so we should use that. */
+        rs_tube_copy_from_scoop(job);
+    }
+    
+    if (job->copy_len) {
+        size_t  this_copy;
+
+        this_copy = rs_buffers_copy(stream, job->copy_len);
+
+        job->copy_len -= this_copy;
+
+        rs_trace("copied %.0f bytes from input buffer, %.0f remain to be copied",
+                 (double) this_copy, (double) job->copy_len);
+    }
+}
+
+
+/*
+ * Put whatever will fit from the tube into the output of the stream.
+ * Return RS_DONE if the tube is now empty and ready to accept another
+ * command, RS_BLOCKED if there is still stuff waiting to go out.
+ */
+int rs_tube_catchup(rs_job_t *job)
+{
+    if (job->write_len)
+        rs_tube_catchup_write(job);
+
+    if (job->write_len) {
+        /* there is still write data queued, so we can't send
+         * anything else. */
+        return RS_BLOCKED;
+    }
+
+    if (job->copy_len)
+        rs_tube_catchup_copy(job);
+    
+    if (job->copy_len) {
+        if (job->stream->eof_in && !job->stream->avail_in && !job->scoop_avail) {
+            rs_log(RS_LOG_ERR,
+                   "reached end of file while copying literal data through buffers");
+            return RS_INPUT_ENDED;
+        }
+
+        return RS_BLOCKED;
+    }
+
+    return RS_DONE;
+}
+
+
+/* Check whether there is data in the tube waiting to go out.  So if true
+ * this basically means that the previous command has finished doing all its
+ * output. */
+int rs_tube_is_idle(rs_job_t const *job)
+{
+    return job->write_len == 0 && job->copy_len == 0;
+}
+
+
+/**
+ * Queue up a request to copy through \p len bytes from the input to
+ * the output of the stream.
+ *
+ * The data is copied from the scoop (if there is anything there) or
+ * from the input, on the next call to rs_tube_write().
+ *
+ * We can only accept this request if there is no copy command already
+ * pending.
+ */
+/* TODO: Try to do the copy immediately, and return a result.  Then,
+ * people can try to continue if possible.  Is this really required?
+ * Callers can just go out and back in again after flushing the
+ * tube. */
+void rs_tube_copy(rs_job_t *job, int len)
+{
+    assert(job->copy_len == 0);
+
+    job->copy_len = len;
+}
+
+
+
+/*
+ * Push some data into the tube for storage.  The tube's never
+ * supposed to get very big, so this will just pop loudly if you do
+ * that.
+ *
+ * We can't accept write data if there's already a copy command in the
+ * tube, because the write data comes out first.
+ */
+void
+rs_tube_write(rs_job_t *job, const void *buf, size_t len)
+{
+    assert(job->copy_len == 0);
+
+    if (len > sizeof(job->write_buf) - job->write_len) {
+        rs_fatal("tube popped when trying to write %ld bytes!",
+                 (long) len);
+    }
+
+    memcpy(job->write_buf + job->write_len, buf, len);
+    job->write_len += len;
+}