wvsync/wvsyncobj.cc

   1 /*
   2  * Worldvisions Weaver Software:
   3  *   Copyright (C) 1997-2004 Net Integration Technologies, Inc.
   4  *
   5  * See wvsyncobj.h for details.
   6  *
   7  */
   8
   9 // Uncomment this to debug...
  10 //#define RSYNC_TRACE 1
  11
  12 #include <string.h>
  13 #include "wvsyncobj.h"
  14
  15 // FIXME: This "C" wrapper must be here until the librsync people fix their
  16 // header file.
  17 extern "C" {
  18     #include <librsync.h>
  19 }
  20
  21
  22 #define CIRCULAR_SIZE   16384       // pretty arbitrary
  23 #define CIRCULAR_THRESH 8192        // should be half of CIRCULAR_SIZE
  24 #define OUTBUF_MAX      1048576     // 1 meg
  25
  26
  27 // given the size of the object we're generating a signature for, produce
  28 // an optimal blocksize.  Currently we use the square root of the
  29 // approximate size of the object, as suggested by the librsync authors.
  30 // This is bounded by 1024 bytes and 65536 bytes, for the sake of sanity.
  31 static size_t optimal_blocksize(off_t approxsize)
  32 {
  33     if (approxsize < 0)
  34         return RS_DEFAULT_BLOCK_LEN;
  35
  36     // do the calculations in divided-by-1024 land, since it's easier.
  37     // dealing with a blocksize that isn't a multiple of 1 kb seems a
  38     // little silly anyway.
  39     approxsize /= 1048576;      // squary squary
  40
  41     size_t k;
  42
  43     if (approxsize <= 1)
  44         k=1;
  45     else if (approxsize >= 4096)
  46         k=64;
  47     else
  48     {
  49         k=0;
  50         size_t thisbit=32;
  51         size_t ksq;
  52         size_t ksqplus;
  53         do {
  54             size_t test = k | thisbit;
  55             if (test*test < approxsize)
  56                 k = test;
  57             ksq = k*k;
  58             ksqplus = (k+1)*(k+1);
  59             if (thisbit == 0)
  60                 break;
  61             thisbit /= 2;
  62         } while (ksq > approxsize || ksqplus <= approxsize);
  63     }
  64
  65     return k * 1024;
  66 }
  67
  68
  69 WvSyncObj::WvSyncObj(WvStringParm _name)
  70 : name(_name), log("WvSyncObj", WvLog::Debug5), err(log.split(WvLog::Error))
  71 {
  72 }
  73
  74
  75 /** jobiter() is used internally.  It's a wrapper for rs_job_iter() to run
  76  * a job with handy debugging output on the buffering, and to print an
  77  * error message when something goes wrong.
  78  */
  79 static rs_result jobiter(WvStringParm id, rs_job_t *job,
  80                          rs_buffers_t *bufs, WvLog &log)
  81 {
  82     rs_result res;
  83
  84 #if RSYNC_TRACE
  85     log("=========== %s\n", id);
  86     log("avail_in=%s eof_in=%s avail_out=%s\n",
  87         bufs->avail_in, bufs->eof_in, bufs->avail_out);
  88 #endif
  89
  90     res = rs_job_iter(job, bufs);
  91
  92 #if RSYNC_TRACE
  93     log("avail_in=%s eof_in=%s avail_out=%s res=%s\n",
  94         bufs->avail_in, bufs->eof_in, bufs->avail_out, res);
  95     log("===========\n\n");
  96 #endif
  97     if (res > 2)                            // see librsync.h
  98         log(WvLog::Error, "Uh oh: %s\n", rs_strerror(res));
  99
 100     return res;
 101 }
 102
 103
 104 /** _jobloop() is used internally, and is a big loop.  If no "inbuf" or
 105  * "infile" is given, it goes through the entire contents of the object
 106  * (using getdata()), runs a given job (using jobiter()) and a
 107  * WvCircularBuf to buffer the input, placing the entire set of results in
 108  * "out".
 109  *
 110  * If an "inbuf" is given, that is used as the input for the job, and
 111  * getdata() is *not* called.
 112  *
 113  * If an "infile" is given, that file is used as the input for the job, and
 114  * getdata() is *not* called.  This is mainly used for input deltas, which
 115  * can be quite large, and might be stored in a file in that case.
 116  */
 117 static rs_result _jobloop(WvStringParm id, rs_job_t *job,
 118                           WvBuf *inbuf, WvFile *infile, WvBuf &out,
 119                           WvSyncObj &obj, WvLog &log, WvSyncCallback cb)
 120 {
 121     rs_buffers_t bufs;
 122     memset(&bufs, '\0', sizeof(rs_buffers_t));
 123
 124     // use a new circular buffer if no "inbuf" was given, and use
 125     // getdata() to fill it as we go (as in getsig and makedelta.)
 126     // If an "inbuf" was given, just use that, and don't call getdata()
 127     // (as in applydelta).
 128     // If an "infile" was given, just use that, and don't call getdata()
 129     // (as in applydelta).
 130     WvCircularBuf circ(CIRCULAR_SIZE);
 131     WvBuf &myinbuf = inbuf ? *inbuf : circ;
 132
 133     rs_result res = RS_DONE;
 134
 135     // loop through the data, calling getdata() if appropriate shoving the
 136     // results into out.
 137     bool done = false;
 138     off_t ofs = 0;
 139
 140     // if we're not calling getdata() or reading from infile, we're already
 141     // done.
 142     if (inbuf)
 143         done = true;
 144
 145     while (myinbuf.used() || !done || res != RS_DONE)
 146     {
 147         size_t avail_in = myinbuf.used();
 148
 149         // get more data?
 150         if (!inbuf && !infile && !done && (avail_in < CIRCULAR_THRESH))
 151         {
 152             done = obj.getdata(myinbuf, ofs, CIRCULAR_THRESH);
 153             ofs += (myinbuf.used() - avail_in);
 154             avail_in = myinbuf.used();
 155         }
 156         else if (infile && !done && (avail_in < CIRCULAR_THRESH))
 157         {
 158             done = !infile->read(myinbuf, CIRCULAR_THRESH);
 159             avail_in = myinbuf.used();
 160         }
 161
 162         // process some data?
 163         bufs.next_in = (char *) myinbuf.get(avail_in);  // might be NULL
 164         bufs.avail_in = avail_in;                       // might be 0
 165         bufs.eof_in = (done && !myinbuf.used());
 166         size_t try_out = CIRCULAR_SIZE;     // FIXME?
 167         bufs.next_out = (char *) out.alloc(try_out);
 168         bufs.avail_out = try_out;
 169
 170         res = jobiter(id, job, &bufs, log);
 171
 172         // we probably didn't use all that buffer space
 173         out.unalloc(bufs.avail_out);
 174         myinbuf.unget(bufs.avail_in);
 175
 176         // is the out buffer getting sorta full?
 177         if (cb && out.used() > OUTBUF_MAX)
 178             cb(obj, out);
 179     }
 180
 181     return res;
 182 }
 183
 184
 185 /** assorted stubs for _jobloop(), for convenience */
 186 static rs_result jobloop(WvStringParm id, rs_job_t *job, WvBuf &out,
 187                          WvSyncObj &obj, WvLog &log, WvSyncCallback cb=0)
 188 {
 189     return _jobloop(id, job, NULL, NULL, out, obj, log, cb);
 190 }
 191
 192
 193 static rs_result jobloop(WvStringParm id, rs_job_t *job,
 194                          WvBuf &inbuf, WvBuf &out,
 195                          WvSyncObj &obj, WvLog &log, WvSyncCallback cb=0)
 196 {
 197     return _jobloop(id, job, &inbuf, NULL, out, obj, log, cb);
 198 }
 199
 200
 201 static rs_result jobloop(WvStringParm id, rs_job_t *job,
 202                          WvStringParm infile, WvBuf &out,
 203                          WvSyncObj &obj, WvLog &log, WvSyncCallback cb=0)
 204 {
 205     WvFile f(infile, O_RDONLY);
 206     if (f.isok())
 207     {
 208         rs_result res = _jobloop(id, job, NULL, &f, out, obj, log, cb);
 209         f.close();
 210         return res;
 211     }
 212     return RS_IO_ERROR;
 213 }
 214
 215
 216 int WvSyncObj::getsig(WvBuf &out)
 217 {
 218     size_t blocksize = optimal_blocksize(approxsize());
 219     rs_job_t *job;
 220     rs_result res;
 221
 222 #if RSYNC_TRACE
 223     log("Using blocksize %s\n", blocksize);
 224 #endif
 225
 226     job = rs_sig_begin(blocksize, RS_DEFAULT_STRONG_LEN);
 227     res = jobloop("sig", job, out, *this, log);
 228     rs_job_free(job);
 229
 230     return res;
 231 }
 232
 233
 234 int WvSyncObj::makedelta(WvBuf &in, WvBuf &out, WvSyncCallback cb)
 235 {
 236     if (in.used() <= 0)
 237     {
 238         err("makedata() called with no signature\n");
 239         return RS_CORRUPT;  // or something
 240     }
 241
 242     rs_job_t *job;
 243     rs_signature_t *somesig;
 244     rs_buffers_t bufs;
 245     rs_result res = RS_DONE;
 246
 247     // must first pull the (remote) signature out of "in", load it into
 248     // "somesig", and build the librsync hash table.
 249     job = rs_loadsig_begin(&somesig);
 250     memset(&bufs, '\0', sizeof(rs_buffers_t));
 251
 252     // it's reasonably easy to load the signature, because we already have
 253     // the entire thing.
 254     size_t sigsize = in.used();
 255     bufs.next_in = (char *) in.get(sigsize);
 256     bufs.avail_in = sigsize;
 257     bufs.eof_in = true;
 258     bufs.next_out = NULL;   // there's no output for loadsig.
 259     bufs.avail_out = 0;
 260     do
 261     {
 262         res = jobiter("loadsig", job, &bufs, log);
 263     }
 264     while( res != RS_DONE );
 265
 266     // clean up that job
 267     rs_job_free(job);
 268
 269     // build the hash table from the signature we just loaded
 270     res = rs_build_hash_table(somesig);
 271     if (res > 2)                            // see librsync.h
 272         err("Error building hash table: %s\n", rs_strerror(res));
 273
 274     // generate the delta and stick it in the "out" buffer.
 275     // we'll do the same circular buffer trick as in getsig(), using
 276     // getdata() the same way.
 277     job = rs_delta_begin(somesig);
 278     res = jobloop("delta", job, out, *this, log, cb);
 279     rs_job_free(job);
 280
 281     return res;
 282 }
 283
 284
 285 // librsync uses this function signature for its callback in the patch job.
 286 // we'll just wrap getdata() inside it.
 287 static rs_result getdata_wrapper(void *arg, rs_long_t pos,
 288                                  size_t *len, void **buf)
 289 {
 290 #if RSYNC_TRACE
 291     printf( "--> getdata_wrapper() called\n" );
 292 #endif
 293
 294     WvSyncObj *obj = (WvSyncObj *)arg;
 295
 296     assert(obj && len && buf && *buf);
 297
 298     WvInPlaceBuf wvbuf(*buf, 0, *len);
 299     bool done = obj->getdata(wvbuf, pos, *len);
 300
 301     if (done)
 302         return RS_INPUT_ENDED;
 303     else
 304     {
 305         *len = wvbuf.used();
 306         return RS_DONE;
 307     }
 308 }
 309
 310
 311 int WvSyncObj::applydelta(WvBuf &in, WvBuf &out, WvSyncCallback cb)
 312 {
 313     rs_job_t *job;
 314     rs_result res;
 315
 316     job = rs_patch_begin(getdata_wrapper, this);
 317     res = jobloop("patch", job, in, out, *this, log, cb);
 318     rs_job_free(job);
 319
 320     return res;
 321 }
 322
 323
 324 int WvSyncObj::applydelta(WvStringParm infile, WvBuf &out, WvSyncCallback cb)
 325 {
 326     rs_job_t *job;
 327     rs_result res;
 328
 329     job = rs_patch_begin(getdata_wrapper, this);
 330     res = jobloop("patch", job, infile, out, *this, log, cb);
 331     rs_job_free(job);
 332
 333     return res;
 334 }
 335
 336
 337 void WvSyncObj::forget(UniConf &cfg)
 338 {
 339     // wipe any "old" information, when it becomes worthless
 340     cfg["old mtime"].setme(WvString::null);
 341     cfg["old meta"].setme(WvString::null);
 342     cfg["old md5sig"].setme(WvString::null);
 343 }
 344
 345
 346 void WvSyncObj::revert(UniConf &cfg)
 347 {
 348     time_t mtime = cfg["old mtime"].getmeint(-1);
 349     WvString meta = cfg["old meta"].getme(WvString::null);
 350     WvString md5sig = cfg["old md5sig"].getme(WvString::null);
 351
 352     if (mtime >= 0)
 353         cfg["mtime"].setmeint(mtime);
 354     if (!!meta)
 355         cfg["meta"].setme(meta);
 356     if (!!md5sig)
 357         cfg["md5sig"].setme(md5sig);
 358 }
 359
 360
 361 bool WvSyncObj::revertible(UniConf &cfg)
 362 {
 363     time_t mtime = cfg["old mtime"].getmeint(-1);
 364     WvString meta = cfg["old meta"].getme(WvString::null);
 365     WvString md5sig = cfg["old md5sig"].getme(WvString::null);
 366
 367     return mtime >= 0 || !!meta || !!md5sig;
 368 }