streaming.c

   1 /*
   2  * Copyright (c) 2011, Google Inc.
   3  */
   4 #include "cache.h"
   5 #include "environment.h"
   6 #include "streaming.h"
   7 #include "repository.h"
   8 #include "object-store.h"
   9 #include "replace-object.h"
  10 #include "packfile.h"
  11 #include "wrapper.h"
  12
  13 typedef int (*open_istream_fn)(struct git_istream *,
  14                                struct repository *,
  15                                const struct object_id *,
  16                                enum object_type *);
  17 typedef int (*close_istream_fn)(struct git_istream *);
  18 typedef ssize_t (*read_istream_fn)(struct git_istream *, char *, size_t);
  19
  20 #define FILTER_BUFFER (1024*16)
  21
  22 struct filtered_istream {
  23         struct git_istream *upstream;
  24         struct stream_filter *filter;
  25         char ibuf[FILTER_BUFFER];
  26         char obuf[FILTER_BUFFER];
  27         int i_end, i_ptr;
  28         int o_end, o_ptr;
  29         int input_finished;
  30 };
  31
  32 struct git_istream {
  33         open_istream_fn open;
  34         close_istream_fn close;
  35         read_istream_fn read;
  36
  37         unsigned long size; /* inflated size of full object */
  38         git_zstream z;
  39         enum { z_unused, z_used, z_done, z_error } z_state;
  40
  41         union {
  42                 struct {
  43                         char *buf; /* from oid_object_info_extended() */
  44                         unsigned long read_ptr;
  45                 } incore;
  46
  47                 struct {
  48                         void *mapped;
  49                         unsigned long mapsize;
  50                         char hdr[32];
  51                         int hdr_avail;
  52                         int hdr_used;
  53                 } loose;
  54
  55                 struct {
  56                         struct packed_git *pack;
  57                         off_t pos;
  58                 } in_pack;
  59
  60                 struct filtered_istream filtered;
  61         } u;
  62 };
  63
  64 /*****************************************************************
  65  *
  66  * Common helpers
  67  *
  68  *****************************************************************/
  69
  70 static void close_deflated_stream(struct git_istream *st)
  71 {
  72         if (st->z_state == z_used)
  73                 git_inflate_end(&st->z);
  74 }
  75
  76
  77 /*****************************************************************
  78  *
  79  * Filtered stream
  80  *
  81  *****************************************************************/
  82
  83 static int close_istream_filtered(struct git_istream *st)
  84 {
  85         free_stream_filter(st->u.filtered.filter);
  86         return close_istream(st->u.filtered.upstream);
  87 }
  88
  89 static ssize_t read_istream_filtered(struct git_istream *st, char *buf,
  90                                      size_t sz)
  91 {
  92         struct filtered_istream *fs = &(st->u.filtered);
  93         size_t filled = 0;
  94
  95         while (sz) {
  96                 /* do we already have filtered output? */
  97                 if (fs->o_ptr < fs->o_end) {
  98                         size_t to_move = fs->o_end - fs->o_ptr;
  99                         if (sz < to_move)
 100                                 to_move = sz;
 101                         memcpy(buf + filled, fs->obuf + fs->o_ptr, to_move);
 102                         fs->o_ptr += to_move;
 103                         sz -= to_move;
 104                         filled += to_move;
 105                         continue;
 106                 }
 107                 fs->o_end = fs->o_ptr = 0;
 108
 109                 /* do we have anything to feed the filter with? */
 110                 if (fs->i_ptr < fs->i_end) {
 111                         size_t to_feed = fs->i_end - fs->i_ptr;
 112                         size_t to_receive = FILTER_BUFFER;
 113                         if (stream_filter(fs->filter,
 114                                           fs->ibuf + fs->i_ptr, &to_feed,
 115                                           fs->obuf, &to_receive))
 116                                 return -1;
 117                         fs->i_ptr = fs->i_end - to_feed;
 118                         fs->o_end = FILTER_BUFFER - to_receive;
 119                         continue;
 120                 }
 121
 122                 /* tell the filter to drain upon no more input */
 123                 if (fs->input_finished) {
 124                         size_t to_receive = FILTER_BUFFER;
 125                         if (stream_filter(fs->filter,
 126                                           NULL, NULL,
 127                                           fs->obuf, &to_receive))
 128                                 return -1;
 129                         fs->o_end = FILTER_BUFFER - to_receive;
 130                         if (!fs->o_end)
 131                                 break;
 132                         continue;
 133                 }
 134                 fs->i_end = fs->i_ptr = 0;
 135
 136                 /* refill the input from the upstream */
 137                 if (!fs->input_finished) {
 138                         fs->i_end = read_istream(fs->upstream, fs->ibuf, FILTER_BUFFER);
 139                         if (fs->i_end < 0)
 140                                 return -1;
 141                         if (fs->i_end)
 142                                 continue;
 143                 }
 144                 fs->input_finished = 1;
 145         }
 146         return filled;
 147 }
 148
 149 static struct git_istream *attach_stream_filter(struct git_istream *st,
 150                                                 struct stream_filter *filter)
 151 {
 152         struct git_istream *ifs = xmalloc(sizeof(*ifs));
 153         struct filtered_istream *fs = &(ifs->u.filtered);
 154
 155         ifs->close = close_istream_filtered;
 156         ifs->read = read_istream_filtered;
 157         fs->upstream = st;
 158         fs->filter = filter;
 159         fs->i_end = fs->i_ptr = 0;
 160         fs->o_end = fs->o_ptr = 0;
 161         fs->input_finished = 0;
 162         ifs->size = -1; /* unknown */
 163         return ifs;
 164 }
 165
 166 /*****************************************************************
 167  *
 168  * Loose object stream
 169  *
 170  *****************************************************************/
 171
 172 static ssize_t read_istream_loose(struct git_istream *st, char *buf, size_t sz)
 173 {
 174         size_t total_read = 0;
 175
 176         switch (st->z_state) {
 177         case z_done:
 178                 return 0;
 179         case z_error:
 180                 return -1;
 181         default:
 182                 break;
 183         }
 184
 185         if (st->u.loose.hdr_used < st->u.loose.hdr_avail) {
 186                 size_t to_copy = st->u.loose.hdr_avail - st->u.loose.hdr_used;
 187                 if (sz < to_copy)
 188                         to_copy = sz;
 189                 memcpy(buf, st->u.loose.hdr + st->u.loose.hdr_used, to_copy);
 190                 st->u.loose.hdr_used += to_copy;
 191                 total_read += to_copy;
 192         }
 193
 194         while (total_read < sz) {
 195                 int status;
 196
 197                 st->z.next_out = (unsigned char *)buf + total_read;
 198                 st->z.avail_out = sz - total_read;
 199                 status = git_inflate(&st->z, Z_FINISH);
 200
 201                 total_read = st->z.next_out - (unsigned char *)buf;
 202
 203                 if (status == Z_STREAM_END) {
 204                         git_inflate_end(&st->z);
 205                         st->z_state = z_done;
 206                         break;
 207                 }
 208                 if (status != Z_OK && (status != Z_BUF_ERROR || total_read < sz)) {
 209                         git_inflate_end(&st->z);
 210                         st->z_state = z_error;
 211                         return -1;
 212                 }
 213         }
 214         return total_read;
 215 }
 216
 217 static int close_istream_loose(struct git_istream *st)
 218 {
 219         close_deflated_stream(st);
 220         munmap(st->u.loose.mapped, st->u.loose.mapsize);
 221         return 0;
 222 }
 223
 224 static int open_istream_loose(struct git_istream *st, struct repository *r,
 225                               const struct object_id *oid,
 226                               enum object_type *type)
 227 {
 228         struct object_info oi = OBJECT_INFO_INIT;
 229         oi.sizep = &st->size;
 230         oi.typep = type;
 231
 232         st->u.loose.mapped = map_loose_object(r, oid, &st->u.loose.mapsize);
 233         if (!st->u.loose.mapped)
 234                 return -1;
 235         switch (unpack_loose_header(&st->z, st->u.loose.mapped,
 236                                     st->u.loose.mapsize, st->u.loose.hdr,
 237                                     sizeof(st->u.loose.hdr), NULL)) {
 238         case ULHR_OK:
 239                 break;
 240         case ULHR_BAD:
 241         case ULHR_TOO_LONG:
 242                 goto error;
 243         }
 244         if (parse_loose_header(st->u.loose.hdr, &oi) < 0 || *type < 0)
 245                 goto error;
 246
 247         st->u.loose.hdr_used = strlen(st->u.loose.hdr) + 1;
 248         st->u.loose.hdr_avail = st->z.total_out;
 249         st->z_state = z_used;
 250         st->close = close_istream_loose;
 251         st->read = read_istream_loose;
 252
 253         return 0;
 254 error:
 255         git_inflate_end(&st->z);
 256         munmap(st->u.loose.mapped, st->u.loose.mapsize);
 257         return -1;
 258 }
 259
 260
 261 /*****************************************************************
 262  *
 263  * Non-delta packed object stream
 264  *
 265  *****************************************************************/
 266
 267 static ssize_t read_istream_pack_non_delta(struct git_istream *st, char *buf,
 268                                            size_t sz)
 269 {
 270         size_t total_read = 0;
 271
 272         switch (st->z_state) {
 273         case z_unused:
 274                 memset(&st->z, 0, sizeof(st->z));
 275                 git_inflate_init(&st->z);
 276                 st->z_state = z_used;
 277                 break;
 278         case z_done:
 279                 return 0;
 280         case z_error:
 281                 return -1;
 282         case z_used:
 283                 break;
 284         }
 285
 286         while (total_read < sz) {
 287                 int status;
 288                 struct pack_window *window = NULL;
 289                 unsigned char *mapped;
 290
 291                 mapped = use_pack(st->u.in_pack.pack, &window,
 292                                   st->u.in_pack.pos, &st->z.avail_in);
 293
 294                 st->z.next_out = (unsigned char *)buf + total_read;
 295                 st->z.avail_out = sz - total_read;
 296                 st->z.next_in = mapped;
 297                 status = git_inflate(&st->z, Z_FINISH);
 298
 299                 st->u.in_pack.pos += st->z.next_in - mapped;
 300                 total_read = st->z.next_out - (unsigned char *)buf;
 301                 unuse_pack(&window);
 302
 303                 if (status == Z_STREAM_END) {
 304                         git_inflate_end(&st->z);
 305                         st->z_state = z_done;
 306                         break;
 307                 }
 308
 309                 /*
 310                  * Unlike the loose object case, we do not have to worry here
 311                  * about running out of input bytes and spinning infinitely. If
 312                  * we get Z_BUF_ERROR due to too few input bytes, then we'll
 313                  * replenish them in the next use_pack() call when we loop. If
 314                  * we truly hit the end of the pack (i.e., because it's corrupt
 315                  * or truncated), then use_pack() catches that and will die().
 316                  */
 317                 if (status != Z_OK && status != Z_BUF_ERROR) {
 318                         git_inflate_end(&st->z);
 319                         st->z_state = z_error;
 320                         return -1;
 321                 }
 322         }
 323         return total_read;
 324 }
 325
 326 static int close_istream_pack_non_delta(struct git_istream *st)
 327 {
 328         close_deflated_stream(st);
 329         return 0;
 330 }
 331
 332 static int open_istream_pack_non_delta(struct git_istream *st,
 333                                        struct repository *r UNUSED,
 334                                        const struct object_id *oid UNUSED,
 335                                        enum object_type *type UNUSED)
 336 {
 337         struct pack_window *window;
 338         enum object_type in_pack_type;
 339
 340         window = NULL;
 341
 342         in_pack_type = unpack_object_header(st->u.in_pack.pack,
 343                                             &window,
 344                                             &st->u.in_pack.pos,
 345                                             &st->size);
 346         unuse_pack(&window);
 347         switch (in_pack_type) {
 348         default:
 349                 return -1; /* we do not do deltas for now */
 350         case OBJ_COMMIT:
 351         case OBJ_TREE:
 352         case OBJ_BLOB:
 353         case OBJ_TAG:
 354                 break;
 355         }
 356         st->z_state = z_unused;
 357         st->close = close_istream_pack_non_delta;
 358         st->read = read_istream_pack_non_delta;
 359
 360         return 0;
 361 }
 362
 363
 364 /*****************************************************************
 365  *
 366  * In-core stream
 367  *
 368  *****************************************************************/
 369
 370 static int close_istream_incore(struct git_istream *st)
 371 {
 372         free(st->u.incore.buf);
 373         return 0;
 374 }
 375
 376 static ssize_t read_istream_incore(struct git_istream *st, char *buf, size_t sz)
 377 {
 378         size_t read_size = sz;
 379         size_t remainder = st->size - st->u.incore.read_ptr;
 380
 381         if (remainder <= read_size)
 382                 read_size = remainder;
 383         if (read_size) {
 384                 memcpy(buf, st->u.incore.buf + st->u.incore.read_ptr, read_size);
 385                 st->u.incore.read_ptr += read_size;
 386         }
 387         return read_size;
 388 }
 389
 390 static int open_istream_incore(struct git_istream *st, struct repository *r,
 391                                const struct object_id *oid, enum object_type *type)
 392 {
 393         struct object_info oi = OBJECT_INFO_INIT;
 394
 395         st->u.incore.read_ptr = 0;
 396         st->close = close_istream_incore;
 397         st->read = read_istream_incore;
 398
 399         oi.typep = type;
 400         oi.sizep = &st->size;
 401         oi.contentp = (void **)&st->u.incore.buf;
 402         return oid_object_info_extended(r, oid, &oi,
 403                                         OBJECT_INFO_DIE_IF_CORRUPT);
 404 }
 405
 406 /*****************************************************************************
 407  * static helpers variables and functions for users of streaming interface
 408  *****************************************************************************/
 409
 410 static int istream_source(struct git_istream *st,
 411                           struct repository *r,
 412                           const struct object_id *oid,
 413                           enum object_type *type)
 414 {
 415         unsigned long size;
 416         int status;
 417         struct object_info oi = OBJECT_INFO_INIT;
 418
 419         oi.typep = type;
 420         oi.sizep = &size;
 421         status = oid_object_info_extended(r, oid, &oi, 0);
 422         if (status < 0)
 423                 return status;
 424
 425         switch (oi.whence) {
 426         case OI_LOOSE:
 427                 st->open = open_istream_loose;
 428                 return 0;
 429         case OI_PACKED:
 430                 if (!oi.u.packed.is_delta && big_file_threshold < size) {
 431                         st->u.in_pack.pack = oi.u.packed.pack;
 432                         st->u.in_pack.pos = oi.u.packed.offset;
 433                         st->open = open_istream_pack_non_delta;
 434                         return 0;
 435                 }
 436                 /* fallthru */
 437         default:
 438                 st->open = open_istream_incore;
 439                 return 0;
 440         }
 441 }
 442
 443 /****************************************************************
 444  * Users of streaming interface
 445  ****************************************************************/
 446
 447 int close_istream(struct git_istream *st)
 448 {
 449         int r = st->close(st);
 450         free(st);
 451         return r;
 452 }
 453
 454 ssize_t read_istream(struct git_istream *st, void *buf, size_t sz)
 455 {
 456         return st->read(st, buf, sz);
 457 }
 458
 459 struct git_istream *open_istream(struct repository *r,
 460                                  const struct object_id *oid,
 461                                  enum object_type *type,
 462                                  unsigned long *size,
 463                                  struct stream_filter *filter)
 464 {
 465         struct git_istream *st = xmalloc(sizeof(*st));
 466         const struct object_id *real = lookup_replace_object(r, oid);
 467         int ret = istream_source(st, r, real, type);
 468
 469         if (ret) {
 470                 free(st);
 471                 return NULL;
 472         }
 473
 474         if (st->open(st, r, real, type)) {
 475                 if (open_istream_incore(st, r, real, type)) {
 476                         free(st);
 477                         return NULL;
 478                 }
 479         }
 480         if (filter) {
 481                 /* Add "&& !is_null_stream_filter(filter)" for performance */
 482                 struct git_istream *nst = attach_stream_filter(st, filter);
 483                 if (!nst) {
 484                         close_istream(st);
 485                         return NULL;
 486                 }
 487                 st = nst;
 488         }
 489
 490         *size = st->size;
 491         return st;
 492 }
 493
 494 int stream_blob_to_fd(int fd, const struct object_id *oid, struct stream_filter *filter,
 495                       int can_seek)
 496 {
 497         struct git_istream *st;
 498         enum object_type type;
 499         unsigned long sz;
 500         ssize_t kept = 0;
 501         int result = -1;
 502
 503         st = open_istream(the_repository, oid, &type, &sz, filter);
 504         if (!st) {
 505                 if (filter)
 506                         free_stream_filter(filter);
 507                 return result;
 508         }
 509         if (type != OBJ_BLOB)
 510                 goto close_and_exit;
 511         for (;;) {
 512                 char buf[1024 * 16];
 513                 ssize_t wrote, holeto;
 514                 ssize_t readlen = read_istream(st, buf, sizeof(buf));
 515
 516                 if (readlen < 0)
 517                         goto close_and_exit;
 518                 if (!readlen)
 519                         break;
 520                 if (can_seek && sizeof(buf) == readlen) {
 521                         for (holeto = 0; holeto < readlen; holeto++)
 522                                 if (buf[holeto])
 523                                         break;
 524                         if (readlen == holeto) {
 525                                 kept += holeto;
 526                                 continue;
 527                         }
 528                 }
 529
 530                 if (kept && lseek(fd, kept, SEEK_CUR) == (off_t) -1)
 531                         goto close_and_exit;
 532                 else
 533                         kept = 0;
 534                 wrote = write_in_full(fd, buf, readlen);
 535
 536                 if (wrote < 0)
 537                         goto close_and_exit;
 538         }
 539         if (kept && (lseek(fd, kept - 1, SEEK_CUR) == (off_t) -1 ||
 540                      xwrite(fd, "", 1) != 1))
 541                 goto close_and_exit;
 542         result = 0;
 543
 544  close_and_exit:
 545         close_istream(st);
 546         return result;
 547 }