| 1 | /* |
| 2 | * Copyright (c) 2011, Google Inc. |
| 3 | */ |
| 4 | |
| 5 | #include "git-compat-util.h" |
| 6 | #include "convert.h" |
| 7 | #include "environment.h" |
| 8 | #include "repository.h" |
| 9 | #include "odb.h" |
| 10 | #include "odb/source.h" |
| 11 | #include "odb/streaming.h" |
| 12 | #include "replace-object.h" |
| 13 | |
| 14 | #define FILTER_BUFFER (1024*16) |
| 15 | |
| 16 | /***************************************************************** |
| 17 | * |
| 18 | * Filtered stream |
| 19 | * |
| 20 | *****************************************************************/ |
| 21 | |
| 22 | struct odb_filtered_read_stream { |
| 23 | struct odb_read_stream base; |
| 24 | struct odb_read_stream *upstream; |
| 25 | struct stream_filter *filter; |
| 26 | char ibuf[FILTER_BUFFER]; |
| 27 | char obuf[FILTER_BUFFER]; |
| 28 | int i_end, i_ptr; |
| 29 | int o_end, o_ptr; |
| 30 | int input_finished; |
| 31 | }; |
| 32 | |
| 33 | static int close_istream_filtered(struct odb_read_stream *_fs) |
| 34 | { |
| 35 | struct odb_filtered_read_stream *fs = (struct odb_filtered_read_stream *)_fs; |
| 36 | free_stream_filter(fs->filter); |
| 37 | return odb_read_stream_close(fs->upstream); |
| 38 | } |
| 39 | |
| 40 | static ssize_t read_istream_filtered(struct odb_read_stream *_fs, char *buf, |
| 41 | size_t sz) |
| 42 | { |
| 43 | struct odb_filtered_read_stream *fs = (struct odb_filtered_read_stream *)_fs; |
| 44 | size_t filled = 0; |
| 45 | |
| 46 | while (sz) { |
| 47 | /* do we already have filtered output? */ |
| 48 | if (fs->o_ptr < fs->o_end) { |
| 49 | size_t to_move = fs->o_end - fs->o_ptr; |
| 50 | if (sz < to_move) |
| 51 | to_move = sz; |
| 52 | memcpy(buf + filled, fs->obuf + fs->o_ptr, to_move); |
| 53 | fs->o_ptr += to_move; |
| 54 | sz -= to_move; |
| 55 | filled += to_move; |
| 56 | continue; |
| 57 | } |
| 58 | fs->o_end = fs->o_ptr = 0; |
| 59 | |
| 60 | /* do we have anything to feed the filter with? */ |
| 61 | if (fs->i_ptr < fs->i_end) { |
| 62 | size_t to_feed = fs->i_end - fs->i_ptr; |
| 63 | size_t to_receive = FILTER_BUFFER; |
| 64 | if (stream_filter(fs->filter, |
| 65 | fs->ibuf + fs->i_ptr, &to_feed, |
| 66 | fs->obuf, &to_receive)) |
| 67 | return -1; |
| 68 | fs->i_ptr = fs->i_end - to_feed; |
| 69 | fs->o_end = FILTER_BUFFER - to_receive; |
| 70 | continue; |
| 71 | } |
| 72 | |
| 73 | /* tell the filter to drain upon no more input */ |
| 74 | if (fs->input_finished) { |
| 75 | size_t to_receive = FILTER_BUFFER; |
| 76 | if (stream_filter(fs->filter, |
| 77 | NULL, NULL, |
| 78 | fs->obuf, &to_receive)) |
| 79 | return -1; |
| 80 | fs->o_end = FILTER_BUFFER - to_receive; |
| 81 | if (!fs->o_end) |
| 82 | break; |
| 83 | continue; |
| 84 | } |
| 85 | fs->i_end = fs->i_ptr = 0; |
| 86 | |
| 87 | /* refill the input from the upstream */ |
| 88 | if (!fs->input_finished) { |
| 89 | fs->i_end = odb_read_stream_read(fs->upstream, fs->ibuf, FILTER_BUFFER); |
| 90 | if (fs->i_end < 0) |
| 91 | return -1; |
| 92 | if (fs->i_end) |
| 93 | continue; |
| 94 | } |
| 95 | fs->input_finished = 1; |
| 96 | } |
| 97 | return filled; |
| 98 | } |
| 99 | |
| 100 | static struct odb_read_stream *attach_stream_filter(struct odb_read_stream *st, |
| 101 | struct stream_filter *filter) |
| 102 | { |
| 103 | struct odb_filtered_read_stream *fs; |
| 104 | |
| 105 | CALLOC_ARRAY(fs, 1); |
| 106 | fs->base.close = close_istream_filtered; |
| 107 | fs->base.read = read_istream_filtered; |
| 108 | fs->upstream = st; |
| 109 | fs->filter = filter; |
| 110 | fs->base.size = -1; /* unknown */ |
| 111 | fs->base.type = st->type; |
| 112 | |
| 113 | return &fs->base; |
| 114 | } |
| 115 | |
| 116 | /***************************************************************** |
| 117 | * |
| 118 | * In-core stream |
| 119 | * |
| 120 | *****************************************************************/ |
| 121 | |
| 122 | struct odb_incore_read_stream { |
| 123 | struct odb_read_stream base; |
| 124 | char *buf; /* from odb_read_object_info_extended() */ |
| 125 | unsigned long read_ptr; |
| 126 | }; |
| 127 | |
| 128 | static int close_istream_incore(struct odb_read_stream *_st) |
| 129 | { |
| 130 | struct odb_incore_read_stream *st = (struct odb_incore_read_stream *)_st; |
| 131 | free(st->buf); |
| 132 | return 0; |
| 133 | } |
| 134 | |
| 135 | static ssize_t read_istream_incore(struct odb_read_stream *_st, char *buf, size_t sz) |
| 136 | { |
| 137 | struct odb_incore_read_stream *st = (struct odb_incore_read_stream *)_st; |
| 138 | size_t read_size = sz; |
| 139 | size_t remainder = st->base.size - st->read_ptr; |
| 140 | |
| 141 | if (remainder <= read_size) |
| 142 | read_size = remainder; |
| 143 | if (read_size) { |
| 144 | memcpy(buf, st->buf + st->read_ptr, read_size); |
| 145 | st->read_ptr += read_size; |
| 146 | } |
| 147 | return read_size; |
| 148 | } |
| 149 | |
| 150 | static int open_istream_incore(struct odb_read_stream **out, |
| 151 | struct object_database *odb, |
| 152 | const struct object_id *oid) |
| 153 | { |
| 154 | struct object_info oi = OBJECT_INFO_INIT; |
| 155 | struct odb_incore_read_stream stream = { |
| 156 | .base.close = close_istream_incore, |
| 157 | .base.read = read_istream_incore, |
| 158 | }; |
| 159 | struct odb_incore_read_stream *st; |
| 160 | int ret; |
| 161 | |
| 162 | oi.typep = &stream.base.type; |
| 163 | oi.sizep = &stream.base.size; |
| 164 | oi.contentp = (void **)&stream.buf; |
| 165 | ret = odb_read_object_info_extended(odb, oid, &oi, |
| 166 | OBJECT_INFO_DIE_IF_CORRUPT); |
| 167 | if (ret) |
| 168 | return ret; |
| 169 | |
| 170 | CALLOC_ARRAY(st, 1); |
| 171 | *st = stream; |
| 172 | *out = &st->base; |
| 173 | |
| 174 | return 0; |
| 175 | } |
| 176 | |
| 177 | /***************************************************************************** |
| 178 | * static helpers variables and functions for users of streaming interface |
| 179 | *****************************************************************************/ |
| 180 | |
| 181 | static int istream_source(struct odb_read_stream **out, |
| 182 | struct object_database *odb, |
| 183 | const struct object_id *oid) |
| 184 | { |
| 185 | struct odb_source *source; |
| 186 | |
| 187 | odb_prepare_alternates(odb); |
| 188 | for (source = odb->sources; source; source = source->next) |
| 189 | if (!odb_source_read_object_stream(out, source, oid)) |
| 190 | return 0; |
| 191 | |
| 192 | return open_istream_incore(out, odb, oid); |
| 193 | } |
| 194 | |
| 195 | /**************************************************************** |
| 196 | * Users of streaming interface |
| 197 | ****************************************************************/ |
| 198 | |
| 199 | int odb_read_stream_close(struct odb_read_stream *st) |
| 200 | { |
| 201 | int r = st->close(st); |
| 202 | free(st); |
| 203 | return r; |
| 204 | } |
| 205 | |
| 206 | ssize_t odb_read_stream_read(struct odb_read_stream *st, void *buf, size_t sz) |
| 207 | { |
| 208 | return st->read(st, buf, sz); |
| 209 | } |
| 210 | |
| 211 | struct odb_read_stream *odb_read_stream_open(struct object_database *odb, |
| 212 | const struct object_id *oid, |
| 213 | struct stream_filter *filter) |
| 214 | { |
| 215 | struct odb_read_stream *st; |
| 216 | const struct object_id *real = lookup_replace_object(odb->repo, oid); |
| 217 | int ret = istream_source(&st, odb, real); |
| 218 | |
| 219 | if (ret) |
| 220 | return NULL; |
| 221 | |
| 222 | if (filter) { |
| 223 | /* Add "&& !is_null_stream_filter(filter)" for performance */ |
| 224 | struct odb_read_stream *nst = attach_stream_filter(st, filter); |
| 225 | if (!nst) { |
| 226 | odb_read_stream_close(st); |
| 227 | return NULL; |
| 228 | } |
| 229 | st = nst; |
| 230 | } |
| 231 | |
| 232 | return st; |
| 233 | } |
| 234 | |
| 235 | ssize_t odb_write_stream_read(struct odb_write_stream *st, void *buf, size_t sz) |
| 236 | { |
| 237 | return st->read(st, buf, sz); |
| 238 | } |
| 239 | |
| 240 | void odb_write_stream_release(struct odb_write_stream *st) |
| 241 | { |
| 242 | free(st->data); |
| 243 | } |
| 244 | |
| 245 | int odb_stream_blob_to_fd(struct object_database *odb, |
| 246 | int fd, |
| 247 | const struct object_id *oid, |
| 248 | struct stream_filter *filter, |
| 249 | int can_seek) |
| 250 | { |
| 251 | struct odb_read_stream *st; |
| 252 | ssize_t kept = 0; |
| 253 | int result = -1; |
| 254 | |
| 255 | st = odb_read_stream_open(odb, oid, filter); |
| 256 | if (!st) { |
| 257 | if (filter) |
| 258 | free_stream_filter(filter); |
| 259 | return result; |
| 260 | } |
| 261 | if (st->type != OBJ_BLOB) |
| 262 | goto close_and_exit; |
| 263 | for (;;) { |
| 264 | char buf[1024 * 16]; |
| 265 | ssize_t wrote, holeto; |
| 266 | ssize_t readlen = odb_read_stream_read(st, buf, sizeof(buf)); |
| 267 | |
| 268 | if (readlen < 0) |
| 269 | goto close_and_exit; |
| 270 | if (!readlen) |
| 271 | break; |
| 272 | if (can_seek && sizeof(buf) == readlen) { |
| 273 | for (holeto = 0; holeto < readlen; holeto++) |
| 274 | if (buf[holeto]) |
| 275 | break; |
| 276 | if (readlen == holeto) { |
| 277 | kept += holeto; |
| 278 | continue; |
| 279 | } |
| 280 | } |
| 281 | |
| 282 | if (kept && lseek(fd, kept, SEEK_CUR) == (off_t) -1) |
| 283 | goto close_and_exit; |
| 284 | else |
| 285 | kept = 0; |
| 286 | wrote = write_in_full(fd, buf, readlen); |
| 287 | |
| 288 | if (wrote < 0) |
| 289 | goto close_and_exit; |
| 290 | } |
| 291 | if (kept && (lseek(fd, kept - 1, SEEK_CUR) == (off_t) -1 || |
| 292 | xwrite(fd, "", 1) != 1)) |
| 293 | goto close_and_exit; |
| 294 | result = 0; |
| 295 | |
| 296 | close_and_exit: |
| 297 | odb_read_stream_close(st); |
| 298 | return result; |
| 299 | } |
| 300 | |
| 301 | struct read_object_fd_data { |
| 302 | int fd; |
| 303 | size_t remaining; |
| 304 | }; |
| 305 | |
| 306 | static ssize_t read_object_fd(struct odb_write_stream *stream, |
| 307 | unsigned char *buf, size_t len) |
| 308 | { |
| 309 | struct read_object_fd_data *data = stream->data; |
| 310 | ssize_t read_result; |
| 311 | size_t count; |
| 312 | |
| 313 | if (stream->is_finished) |
| 314 | return 0; |
| 315 | |
| 316 | count = data->remaining < len ? data->remaining : len; |
| 317 | read_result = read_in_full(data->fd, buf, count); |
| 318 | if (read_result < 0 || (size_t)read_result != count) |
| 319 | return -1; |
| 320 | |
| 321 | data->remaining -= count; |
| 322 | if (!data->remaining) |
| 323 | stream->is_finished = 1; |
| 324 | |
| 325 | return read_result; |
| 326 | } |
| 327 | |
| 328 | void odb_write_stream_from_fd(struct odb_write_stream *stream, int fd, |
| 329 | size_t size) |
| 330 | { |
| 331 | struct read_object_fd_data *data; |
| 332 | |
| 333 | CALLOC_ARRAY(data, 1); |
| 334 | data->fd = fd; |
| 335 | data->remaining = size; |
| 336 | |
| 337 | stream->data = data; |
| 338 | stream->read = read_object_fd; |
| 339 | stream->is_finished = 0; |
| 340 | } |