example/eblob_backend.c

   1 /*
   2  * 2008+ Copyright (c) Evgeniy Polyakov <zbr@ioremap.net>
   3  * All rights reserved.
   4  *
   5  * This program is free software; you can redistribute it and/or modify
   6  * it under the terms of the GNU General Public License as published by
   7  * the Free Software Foundation; either version 2 of the License, or
   8  * (at your option) any later version.
   9  *
  10  * This program is distributed in the hope that it will be useful,
  11  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  12  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  13  * GNU General Public License for more details.
  14  */
  15
  16 #define _XOPEN_SOURCE 600
  17
  18 #include <sys/types.h>
  19 #include <sys/stat.h>
  20 #include <sys/socket.h>
  21 #include <sys/mman.h>
  22 #include <sys/wait.h>
  23
  24 #include <errno.h>
  25 #include <ctype.h>
  26 #include <dirent.h>
  27 #include <fcntl.h>
  28 #include <pthread.h>
  29 #include <stdio.h>
  30 #include <stdlib.h>
  31 #include <string.h>
  32 #include <unistd.h>
  33
  34 #include <eblob/blob.h>
  35
  36 #include "elliptics/packet.h"
  37 #include "elliptics/interface.h"
  38
  39 #include "backends.h"
  40 #include "common.h"
  41
  42 #ifndef __unused
  43 #define __unused        __attribute__ ((unused))
  44 #endif
  45
  46 struct eblob_backend_config {
  47         struct eblob_config             data;
  48         struct eblob_backend            *eblob;
  49 };
  50
  51 #if EBLOB_ID_SIZE != DNET_ID_SIZE
  52 #error "EBLOB_ID_SIZE must be equal to DNET_ID_SIZE"
  53 #endif
  54
  55 static int blob_write(struct eblob_backend_config *c, void *state __unused, struct dnet_cmd *cmd __unused, void *data)
  56 {
  57         int err;
  58         struct dnet_io_attr *io = data;
  59         struct eblob_write_control wc;
  60         struct eblob_key key;
  61         uint64_t flags = BLOB_DISK_CTL_WRITE_RETURN;
  62
  63         dnet_backend_log(DNET_LOG_NOTICE, "%s: EBLOB: blob-write: WRITE: start: offset: %llu, size: %llu, ioflags: %x, type: %d.\n",
  64                 dnet_dump_id_str(io->id), (unsigned long long)io->offset, (unsigned long long)io->size, io->flags, io->type);
  65
  66         memset(&wc, 0, sizeof(struct eblob_write_control));
  67         wc.data_fd = -1;
  68
  69         dnet_convert_io_attr(io);
  70
  71         data += sizeof(struct dnet_io_attr);
  72
  73         if (io->flags & DNET_IO_FLAGS_COMPRESS)
  74                 flags |= BLOB_DISK_CTL_COMPRESS;
  75
  76         if (io->flags & DNET_IO_FLAGS_APPEND)
  77                 flags |= BLOB_DISK_CTL_APPEND;
  78
  79         if (io->flags & DNET_IO_FLAGS_OVERWRITE)
  80                 flags |= BLOB_DISK_CTL_OVERWRITE;
  81
  82         if (io->flags & DNET_IO_FLAGS_NOCSUM)
  83                 flags |= BLOB_DISK_CTL_NOCSUM;
  84
  85         memcpy(key.id, io->id, EBLOB_ID_SIZE);
  86
  87         if ((io->type == EBLOB_TYPE_META) && !(io->flags & DNET_IO_FLAGS_META)) {
  88                 dnet_backend_log(DNET_LOG_ERROR, "%s: EBLOB: blob-write: meta-check: COLUMN %d IS RESERVED FOR METADATA\n",
  89                         dnet_dump_id_str(io->id), io->type);
  90                 err = -EPERM;
  91                 goto err_out_exit;
  92         }
  93
  94         if (io->flags & DNET_IO_FLAGS_PREPARE) {
  95                 wc.offset = 0;
  96                 wc.size = io->num;
  97                 wc.flags = flags;
  98                 wc.type = io->type;
  99
 100                 err = eblob_write_prepare(c->eblob, &key, &wc);
 101                 if (err) {
 102                         dnet_backend_log(DNET_LOG_ERROR, "%s: EBLOB: blob-write: eblob_write_prepare: size: %llu: type: %d: %s %d\n",
 103                                 dnet_dump_id_str(io->id), (unsigned long long)io->num, io->type, strerror(-err), err);
 104                         goto err_out_exit;
 105                 }
 106
 107                 dnet_backend_log(DNET_LOG_NOTICE, "%s: EBLOB: blob-write: eblob_write_prepare: size: %llu: type: %d: Ok\n",
 108                         dnet_dump_id_str(io->id), (unsigned long long)io->num, io->type);
 109         }
 110
 111         if (io->size) {
 112                 if (io->flags & DNET_IO_FLAGS_PLAIN_WRITE) {
 113                         err = eblob_plain_write(c->eblob, &key, data, io->offset, io->size, io->type);
 114                 } else {
 115                         err = eblob_write(c->eblob, &key, data, io->offset, io->size, flags, io->type);
 116                         if (!err) {
 117                                 if (io->size >= sizeof(struct eblob_write_control)) {
 118                                         memcpy(&wc, data, sizeof(struct eblob_write_control));
 119                                 }
 120                         }
 121                 }
 122
 123                 if (err) {
 124                         dnet_backend_log(DNET_LOG_ERROR, "%s: EBLOB: blob-write: WRITE: %d: %s\n",
 125                                 dnet_dump_id_str(io->id), err, strerror(-err));
 126                         goto err_out_exit;
 127                 }
 128
 129                 dnet_backend_log(DNET_LOG_NOTICE, "%s: EBLOB: blob-write: WRITE: Ok: offset: %llu, size: %llu, type: %d.\n",
 130                         dnet_dump_id_str(io->id), (unsigned long long)io->offset, (unsigned long long)io->size, io->type);
 131         }
 132
 133         if (io->flags & DNET_IO_FLAGS_COMMIT) {
 134                 wc.offset = 0;
 135                 wc.size = io->num;
 136                 wc.flags = flags;
 137                 wc.type = io->type;
 138
 139                 err = eblob_write_commit(c->eblob, &key, NULL, 0, &wc);
 140                 if (err) {
 141                         dnet_backend_log(DNET_LOG_ERROR, "%s: EBLOB: blob-write: eblob_write_commit: size: %llu: type: %d: %s %d\n",
 142                                 dnet_dump_id_str(io->id), (unsigned long long)io->num, io->type, strerror(-err), err);
 143                         goto err_out_exit;
 144                 }
 145
 146                 dnet_backend_log(DNET_LOG_NOTICE, "%s: EBLOB: blob-write: eblob_write_commit: size: %llu: type: %d: Ok\n",
 147                         dnet_dump_id_str(io->id), (unsigned long long)io->num, io->type);
 148         }
 149
 150         if (!err && wc.data_fd == -1) {
 151                 err = eblob_read_nocsum(c->eblob, &key, &wc.data_fd, &wc.offset, &wc.size, io->type);
 152                 if (err < 0) {
 153                         dnet_backend_log(DNET_LOG_ERROR, "%s: EBLOB: blob-write: eblob_read: "
 154                                         "size: %llu: type: %d: %s %d\n",
 155                                 dnet_dump_id_str(io->id), (unsigned long long)io->num, io->type, strerror(-err), err);
 156                         goto err_out_exit;
 157                 }
 158
 159                 /* data is compressed, but we only care about header */
 160                 if (err == 1) {
 161                         err = 0;
 162                 }
 163         }
 164
 165         err = dnet_send_file_info(state, cmd, wc.data_fd, wc.offset, wc.size);
 166         if (err) {
 167                 dnet_backend_log(DNET_LOG_ERROR, "%s: EBLOB: blob-write: dnet_send_file_info: "
 168                                 "fd: %d, offset: %llu, size: %llu: type: %d: %s %d\n",
 169                         dnet_dump_id_str(io->id), wc.data_fd,(unsigned long long)wc.offset,
 170                         (unsigned long long)wc.size, io->type, strerror(-err), err);
 171                 goto err_out_exit;
 172         }
 173
 174 err_out_exit:
 175         return err;
 176 }
 177
 178 static int blob_read(struct eblob_backend_config *c, void *state, struct dnet_cmd *cmd, void *data)
 179 {
 180         struct dnet_io_attr *io = data;
 181         struct eblob_backend *b = c->eblob;
 182         uint64_t offset, size, orig_offset, orig_size;
 183         struct eblob_key key;
 184         char *read_data = NULL;
 185         int fd, err;
 186
 187         dnet_convert_io_attr(io);
 188
 189         memcpy(key.id, io->id, EBLOB_ID_SIZE);
 190
 191         if (io->flags & DNET_IO_FLAGS_NOCSUM) {
 192                 err = eblob_read_nocsum(b, &key, &fd, &orig_offset, &orig_size, io->type);
 193         } else {
 194                 err = eblob_read(b, &key, &fd, &orig_offset, &orig_size, io->type);
 195         }
 196
 197         if (err < 0) {
 198                 dnet_backend_log(DNET_LOG_ERROR, "%s: EBLOB: blob-read-fd: READ: %d: %s\n",
 199                         dnet_dump_id_str(io->id), err, strerror(-err));
 200                 goto err_out_exit;
 201         } else if (err > 0) {
 202                 /* data is compressed */
 203
 204                 size = 0;
 205                 err = eblob_read_data(b, &key, io->offset, &read_data, &size, io->type);
 206                 if (err) {
 207                         dnet_backend_log(DNET_LOG_ERROR, "%s: EBLOB: blob-read-data: READ: %d: %s\n",
 208                                 dnet_dump_id_str(io->id), err, strerror(-err));
 209                         goto err_out_exit;
 210                 }
 211
 212                 offset = 0; /* to shut up compiler - offset is not used when there is data */
 213                 fd = -1;
 214         } else {
 215                 if (io->offset >= orig_size) {
 216                         err = -E2BIG;
 217                         goto err_out_exit;
 218                 }
 219
 220                 offset = orig_offset + io->offset;
 221                 size = orig_size - io->offset;
 222
 223                 if (io->size && size > io->size)
 224                         size = io->size;
 225         }
 226
 227         io->size = size;
 228         err = dnet_send_read_data(state, cmd, io, read_data, fd, offset, 0);
 229
 230         /* free compressed data */
 231         free(read_data);
 232 err_out_exit:
 233         return err;
 234 }
 235
 236 struct eblob_read_range_priv {
 237         void                    *state;
 238         struct dnet_cmd         *cmd;
 239         struct eblob_range_request      *keys;
 240         uint64_t                keys_size;
 241         uint64_t                keys_cnt;
 242         uint32_t                flags;
 243 };
 244
 245 static int blob_cmp_range_request(const void *req1, const void *req2)
 246 {
 247         return memcmp(((struct eblob_range_request *)(req1))->record_key, ((struct eblob_range_request *)(req2))->record_key, EBLOB_ID_SIZE);
 248 }
 249
 250 static int blob_read_range_callback(struct eblob_range_request *req)
 251 {
 252         struct eblob_read_range_priv *p = req->priv;
 253         struct dnet_io_attr io;
 254         int err;
 255
 256         if (req->requested_offset > req->record_size) {
 257                 err = 0;
 258                 goto err_out_exit;
 259         }
 260
 261         if (!(p->flags & DNET_IO_FLAGS_NODATA)) {
 262                 io.flags = 0;
 263                 io.size = req->record_size - req->requested_offset;
 264                 io.offset = req->requested_offset;
 265                 io.type = req->requested_type;
 266
 267                 memcpy(io.id, req->record_key, DNET_ID_SIZE);
 268                 memcpy(io.parent, req->end, DNET_ID_SIZE);
 269
 270                 err = dnet_send_read_data(p->state, p->cmd, &io, NULL, req->record_fd,
 271                                 req->record_offset + req->requested_offset, 0);
 272                 if (!err)
 273                         req->current_pos++;
 274         } else {
 275                 req->current_pos++;
 276                 err = 0;
 277         }
 278
 279 err_out_exit:
 280         return err;
 281 }
 282
 283 static int blob_del_range_callback(struct eblob_backend_config *c, struct dnet_io_attr *io, struct eblob_range_request *req)
 284 {
 285         //struct eblob_read_range_priv *p = req->priv;
 286         struct eblob_key key;
 287         int err;
 288
 289         dnet_backend_log(DNET_LOG_DEBUG, "%s: EBLOB: blob-read-range: DEL\n",dnet_dump_id_str(req->record_key));
 290         memcpy(key.id, req->record_key, EBLOB_ID_SIZE);
 291         err = eblob_remove(c->eblob, &key, io->type);
 292         if (err) {
 293                 dnet_backend_log(DNET_LOG_DEBUG, "%s: EBLOB: blob-read-range: DEL: err: %d\n",dnet_dump_id_str(req->record_key), err);
 294         }
 295
 296         return err;
 297 }
 298
 299 static int blob_range_callback(struct eblob_range_request *req)
 300 {
 301         struct eblob_read_range_priv *p = req->priv;
 302         int len = 10;
 303         char start_id[len*2+1], end_id[len*2+1], cur_id[2*len+1];
 304         int err = 0;
 305
 306         dnet_dump_id_len_raw(req->start, len, start_id);
 307         dnet_dump_id_len_raw(req->end, len, end_id);
 308         dnet_dump_id_len_raw(req->record_key, len, cur_id);
 309
 310         dnet_backend_log(DNET_LOG_NOTICE, "%s: EBLOB: blob-range: limit: %llu [%llu, %llu]: "
 311                         "start: %s, end: %s: io record/requested: offset: %llu/%llu, size: %llu/%llu, type: %d\n",
 312                         cur_id,
 313                         (unsigned long long)req->current_pos,
 314                         (unsigned long long)req->requested_limit_start, (unsigned long long)req->requested_limit_num,
 315                         start_id, end_id,
 316                         (unsigned long long)req->record_offset, (unsigned long long)req->requested_offset,
 317                         (unsigned long long)req->record_size, (unsigned long long)req->requested_size,
 318                         req->requested_type);
 319
 320         if (req->requested_offset > req->record_size) {
 321                 err = 0;
 322                 goto err_out_exit;
 323         }
 324
 325         if (!(p->keys)) {
 326                 p->keys = (struct eblob_range_request*)malloc(sizeof(struct eblob_range_request) * 1000);
 327                 if (!(p->keys)) {
 328                         err = -ENOMEM;
 329                         dnet_backend_log(DNET_LOG_ERROR, "%s: EBLOB: blob-del-range: can't allocate memory\n", cur_id);
 330                         goto err_out_exit;
 331                 }
 332                 p->keys_size = 1000;
 333         }
 334
 335         if (p->keys_size == p->keys_cnt) {
 336                 p->keys = (struct eblob_range_request*)realloc(p->keys, sizeof(struct eblob_range_request) * p->keys_size * 2);
 337                 if (!(p->keys)) {
 338                         err = -ENOMEM;
 339                         dnet_backend_log(DNET_LOG_ERROR, "%s: EBLOB: blob-del-range: can't re-allocate memory, new size: %llu\n",
 340                                         cur_id, (unsigned long long)(p->keys_size * 2));
 341                         goto err_out_exit;
 342                 }
 343                 p->keys_size *= 2;
 344         }
 345
 346         memcpy(&p->keys[p->keys_cnt], req, sizeof(struct eblob_range_request));
 347         dnet_dump_id_len_raw(p->keys[p->keys_cnt].record_key, len, cur_id);
 348         dnet_backend_log(DNET_LOG_DEBUG, "%s: count: %llu\n", cur_id, (unsigned long long)(p->keys_cnt));
 349         p->keys_cnt++;
 350
 351         if (!err)
 352                 req->current_pos++;
 353 err_out_exit:
 354         return err;
 355 }
 356
 357 static int blob_read_range(struct eblob_backend_config *c, void *state, struct dnet_cmd *cmd, void *data)
 358 {
 359         struct eblob_read_range_priv p;
 360         struct dnet_io_attr *io = data;
 361         struct eblob_backend *b = c->eblob;
 362         struct eblob_range_request req;
 363         uint64_t i, start_from = 0;
 364         int err;
 365
 366         memset(&p, 0, sizeof(p));
 367
 368         p.cmd = cmd;
 369         p.state = state;
 370         p.keys = NULL;
 371         p.keys_size= 0;
 372         p.keys_cnt = 0;
 373         p.flags = io->flags;
 374
 375         dnet_convert_io_attr(io);
 376
 377         memset(&req, 0, sizeof(req));
 378
 379         memcpy(req.start, io->id, EBLOB_ID_SIZE);
 380         memcpy(req.end, io->parent, EBLOB_ID_SIZE);
 381         req.requested_offset = io->offset;
 382         req.requested_size = io->size;
 383         req.requested_limit_start = 0;
 384         req.requested_limit_num = ~0ULL;
 385         req.requested_type = io->type;
 386
 387         req.callback = blob_range_callback;
 388         req.back = b;
 389         req.priv = &p;
 390
 391         err = eblob_read_range(&req);
 392         if (err) {
 393                 dnet_backend_log(DNET_LOG_ERROR, "%s: EBLOB: blob-read-range: %d: %s\n",
 394                         dnet_dump_id_str(io->id), err, strerror(-err));
 395                 goto err_out_exit;
 396         }
 397
 398         if ((cmd->cmd == DNET_CMD_READ_RANGE) && (cmd->flags & DNET_ATTR_SORT)) {
 399                 dnet_backend_log(DNET_LOG_DEBUG, "Sorting keys before sending\n");
 400                 qsort(p.keys, p.keys_cnt, sizeof(struct eblob_range_request), &blob_cmp_range_request);
 401         }
 402
 403         if (cmd->cmd == DNET_CMD_READ_RANGE) {
 404                 start_from = io->start;
 405         }
 406
 407         for (i = start_from; i < p.keys_cnt; ++i) {
 408                 switch(cmd->cmd) {
 409                         case DNET_CMD_READ_RANGE:
 410                                 if ((io->num > 0) && (i >= (io->num + start_from)))
 411                                         break;
 412                                 dnet_backend_log(DNET_LOG_DEBUG, "%s: EBLOB: blob-read-range: READ\n",dnet_dump_id_str(p.keys[i].record_key));
 413                                 err = blob_read_range_callback(&p.keys[i]);
 414                                 break;
 415                         case DNET_CMD_DEL_RANGE:
 416                                 dnet_backend_log(DNET_LOG_DEBUG, "%s: EBLOB: blob-read-range: DEL\n",dnet_dump_id_str(p.keys[i].record_key));
 417                                 err = blob_del_range_callback(c, io, &p.keys[i]);
 418                                 break;
 419                 }
 420
 421                 if (err) {
 422                         dnet_backend_log(DNET_LOG_DEBUG, "%s: EBLOB: blob-read-range: err: %d\n",dnet_dump_id_str(p.keys[i].record_key), err);
 423                         goto err_out_exit;
 424                 }
 425         }
 426
 427         if (req.current_pos) {
 428                 struct dnet_io_attr r;
 429
 430                 memcpy(&r, io, sizeof(struct dnet_io_attr));
 431                 r.num = req.current_pos - start_from;
 432                 r.offset = r.size = 0;
 433
 434                 err = dnet_send_read_data(state, cmd, &r, NULL, -1, 0, 0);
 435         }
 436
 437 err_out_exit:
 438         if (p.keys)
 439                 free(p.keys);
 440
 441         return err;
 442 }
 443
 444 static int blob_del(struct eblob_backend_config *c, struct dnet_cmd *cmd)
 445 {
 446         struct eblob_key key;
 447         int err;
 448
 449         memcpy(key.id, cmd->id.id, EBLOB_ID_SIZE);
 450
 451         if (cmd->id.type != -1) {
 452                 err = eblob_remove(c->eblob, &key, cmd->id.type);
 453         } else {
 454                 err = eblob_remove_all(c->eblob, &key);
 455         }
 456
 457         if (err) {
 458                 dnet_backend_log(DNET_LOG_ERROR, "%s: EBLOB: blob-del: REMOVE: type: %d: %d: %s\n",
 459                         dnet_dump_id_str(cmd->id.id), cmd->id.type, err, strerror(-err));
 460         }
 461
 462         return err;
 463 }
 464
 465 static int eblob_send(void *state, void *priv, struct dnet_id *id)
 466 {
 467         struct dnet_node *n = dnet_get_node_from_state(state);
 468         struct eblob_backend_config *c = priv;
 469         struct eblob_backend *b = c->eblob;
 470         uint64_t offset, size;
 471         struct eblob_key key;
 472         int *types, types_num, i;
 473         int err, fd, ret;
 474
 475         memcpy(key.id, id->id, EBLOB_ID_SIZE);
 476
 477         if (id->type == -1) {
 478                 types_num = eblob_get_types(b, &types);
 479                 if (types_num < 0) {
 480                         err = types_num;
 481                         goto err_out_exit;
 482                 }
 483         } else {
 484                 types_num = 1;
 485                 types = &id->type;
 486         }
 487
 488         err = -ENOENT;
 489         for (i = 0; i < types_num; ++i) {
 490                 if (types[i] == EBLOB_TYPE_META)
 491                         continue;
 492
 493                 dnet_backend_log(DNET_LOG_DEBUG, "trying to send type %d\n", types[i]);
 494                 ret = eblob_read(b, &key, &fd, &offset, &size, types[i]);
 495                 if (ret >= 0) {
 496                         struct dnet_io_control ctl;
 497                         void *result = NULL;
 498
 499                         memset(&ctl, 0, sizeof(ctl));
 500
 501                         ctl.fd = fd;
 502                         ctl.local_offset = offset;
 503
 504                         memcpy(&ctl.id, id, sizeof(struct dnet_id));
 505                         ctl.id.type = types[i];
 506
 507                         ctl.io.offset = 0;
 508                         ctl.io.size = size;
 509                         ctl.io.type = types[i];
 510                         ctl.io.flags = 0;
 511
 512                         struct dnet_session *s = dnet_session_create(n);
 513                         dnet_session_set_groups(s, (int *)&id->group_id, 1);
 514
 515                         err = dnet_write_data_wait(s, &ctl, &result);
 516                         if (err < 0) {
 517                                 goto err_out_free;
 518                         }
 519                         free(result);
 520                         err = 0;
 521                 }
 522         }
 523
 524 err_out_free:
 525         if (id->type == -1)
 526                 free(types);
 527 err_out_exit:
 528         return err;
 529 }
 530
 531 static int blob_file_info(struct eblob_backend_config *c, void *state, struct dnet_cmd *cmd)
 532 {
 533         struct eblob_backend *b = c->eblob;
 534         struct eblob_key key;
 535         uint64_t offset, size;
 536         int fd, err;
 537
 538         memcpy(key.id, cmd->id.id, EBLOB_ID_SIZE);
 539         err = eblob_read(b, &key, &fd, &offset, &size, cmd->id.type);
 540         if (err < 0) {
 541                 dnet_backend_log(DNET_LOG_ERROR, "%s: EBLOB: blob-file-info: info-read: %d: %s.\n",
 542                                 dnet_dump_id(&cmd->id), err, strerror(-err));
 543                 goto err_out_exit;
 544         }
 545
 546         if (size == 0) {
 547                 err = -ENOENT;
 548                 dnet_backend_log(DNET_LOG_INFO, "%s: EBLOB: blob-file-info: info-read: ZERO-SIZE-FILE.\n",
 549                                 dnet_dump_id(&cmd->id));
 550                 goto err_out_exit;
 551         }
 552
 553         err = dnet_send_file_info(state, cmd, fd, offset, size);
 554
 555 err_out_exit:
 556         return err;
 557 }
 558
 559 static int blob_bulk_read(struct eblob_backend_config *c, void *state, struct dnet_cmd *cmd, void *data)
 560 {
 561         int err = -1, ret;
 562         struct dnet_io_attr *io = data;
 563         struct dnet_io_attr *ios = io+1;
 564         uint64_t count = 0;
 565         uint64_t i;
 566
 567         dnet_convert_io_attr(io);
 568         count = io->size / sizeof(struct dnet_io_attr);
 569
 570         for (i = 0; i < count; i++) {
 571                 ret = blob_read(c, state, cmd, &ios[i]);
 572                 if (!ret)
 573                         err = 0;
 574                 else if (err == -1)
 575                         err = ret;
 576         }
 577
 578         return err;
 579 }
 580
 581 static int eblob_backend_checksum(struct dnet_node *n, void *priv, struct dnet_id *id, void *csum, int *csize) {
 582         struct eblob_backend_config *c = priv;
 583         struct eblob_backend *b = c->eblob;
 584         uint64_t offset, size;
 585         struct eblob_key key;
 586         int fd, err;
 587
 588         memcpy(key.id, id->id, EBLOB_ID_SIZE);
 589         err = eblob_read(b, &key, &fd, &offset, &size, EBLOB_TYPE_DATA);
 590         if (err < 0) {
 591                 dnet_backend_log(DNET_LOG_ERROR, "%s: EBLOB: blob-checksum: read: type: %d: %d: %s.\n",
 592                                                         dnet_dump_id_str(id->id), id->type, err, strerror(-err));
 593                 goto err_out_exit;
 594         }
 595         err = 0;
 596         if (!size)
 597                 memset(csum, 0, *csize);
 598         else
 599                 err = dnet_checksum_fd(n, csum, csize, fd, offset, size);
 600
 601 err_out_exit:
 602         return err;
 603 }
 604
 605 static int blob_start_defrag(struct eblob_backend_config *c)
 606 {
 607         return eblob_start_defrag(c->eblob);
 608 }
 609
 610 static int eblob_backend_command_handler(void *state, void *priv, struct dnet_cmd *cmd, void *data)
 611 {
 612         int err;
 613         struct eblob_backend_config *c = priv;
 614         char *path, *p;
 615
 616         switch (cmd->cmd) {
 617                 case DNET_CMD_LOOKUP:
 618                         err = blob_file_info(c, state, cmd);
 619                         break;
 620                 case DNET_CMD_WRITE:
 621                         err = blob_write(c, state, cmd, data);
 622                         break;
 623                 case DNET_CMD_READ:
 624                         err = blob_read(c, state, cmd, data);
 625                         break;
 626                 case DNET_CMD_READ_RANGE:
 627                 case DNET_CMD_DEL_RANGE:
 628                         err = blob_read_range(c, state, cmd, data);
 629                         break;
 630                 case DNET_CMD_STAT:
 631                         path = strdup(c->data.file);
 632                         if (!path) {
 633                                 err = -ENOMEM;
 634                                 break;
 635                         }
 636
 637                         p = strrchr(path, '/');
 638                         if (p) {
 639                                 *p = '\0';
 640                         } else {
 641                                 free(path);
 642                                 path = NULL;
 643                         }
 644
 645                         err = backend_stat(state, path, cmd);
 646                         free(path);
 647                         break;
 648                 case DNET_CMD_DEL:
 649                         err = blob_del(c, cmd);
 650                         break;
 651                 case DNET_CMD_BULK_READ:
 652                         err = blob_bulk_read(c, state, cmd, data);
 653                         break;
 654                 case DNET_CMD_DEFRAG:
 655                         err = blob_start_defrag(c);
 656                         break;
 657                 default:
 658                         err = -EINVAL;
 659                         break;
 660         }
 661
 662         return err;
 663 }
 664
 665 static int dnet_blob_set_sync(struct dnet_config_backend *b, char *key __unused, char *value)
 666 {
 667         struct eblob_backend_config *c = b->data;
 668
 669         c->data.sync = atoi(value);
 670         return 0;
 671 }
 672
 673 static int dnet_blob_set_data(struct dnet_config_backend *b, char *key __unused, char *file)
 674 {
 675         struct eblob_backend_config *c = b->data;
 676         int err;
 677
 678         err = backend_storage_size(b, file);
 679         if (err) {
 680                 char root[strlen(file)+1], *ptr;
 681
 682                 snprintf(root, sizeof(root), "%s", file);
 683                 ptr = strrchr(root, '/');
 684                 if (ptr) {
 685                         *ptr = '\0';
 686                         err = backend_storage_size(b, root);
 687                 }
 688
 689                 if (err)
 690                         return err;
 691         }
 692
 693         free(c->data.file);
 694         c->data.file = strdup(file);
 695         if (!c->data.file)
 696                 return -ENOMEM;
 697
 698         return 0;
 699 }
 700
 701 static int dnet_blob_set_block_size(struct dnet_config_backend *b, char *key __unused, char *value)
 702 {
 703         struct eblob_backend_config *c = b->data;
 704
 705         c->data.bsize = strtoul(value, NULL, 0);
 706         return 0;
 707 }
 708
 709 static int dnet_blob_set_blob_size(struct dnet_config_backend *b, char *key __unused, char *value)
 710 {
 711         struct eblob_backend_config *c = b->data;
 712         uint64_t val = strtoul(value, NULL, 0);
 713
 714         if (strchr(value, 'T'))
 715                 val *= 1024*1024*1024*1024ULL;
 716         else if (strchr(value, 'G'))
 717                 val *= 1024*1024*1024ULL;
 718         else if (strchr(value, 'M'))
 719                 val *= 1024*1024;
 720         else if (strchr(value, 'K'))
 721                 val *= 1024;
 722
 723         c->data.blob_size = val;
 724         return 0;
 725 }
 726
 727 static int dnet_blob_set_records_in_blob(struct dnet_config_backend *b, char *key __unused, char *value)
 728 {
 729         struct eblob_backend_config *c = b->data;
 730         uint64_t val = strtoul(value, NULL, 0);
 731
 732         c->data.records_in_blob = val;
 733         return 0;
 734 }
 735
 736 static int dnet_blob_set_blob_cache_size(struct dnet_config_backend *b, char *key __unused, char *value)
 737 {
 738         struct eblob_backend_config *c = b->data;
 739
 740         c->data.cache_size = strtoul(value, NULL, 0);
 741         return 0;
 742 }
 743
 744 static int dnet_blob_set_defrag_timeout(struct dnet_config_backend *b, char *key __unused, char *value)
 745 {
 746         struct eblob_backend_config *c = b->data;
 747
 748         c->data.defrag_timeout = strtoul(value, NULL, 0);
 749         return 0;
 750 }
 751
 752 static int dnet_blob_set_defrag_percentage(struct dnet_config_backend *b, char *key __unused, char *value)
 753 {
 754         struct eblob_backend_config *c = b->data;
 755
 756         c->data.defrag_percentage = strtoul(value, NULL, 0);
 757         return 0;
 758 }
 759
 760 static int dnet_blob_set_iterate_thread_num(struct dnet_config_backend *b, char *key __unused, char *value)
 761 {
 762         struct eblob_backend_config *c = b->data;
 763
 764         c->data.iterate_threads = strtoul(value, NULL, 0);
 765         return 0;
 766 }
 767
 768 static int dnet_blob_set_blob_flags(struct dnet_config_backend *b, char *key __unused, char *value)
 769 {
 770         struct eblob_backend_config *c = b->data;
 771
 772         c->data.blob_flags = strtoul(value, NULL, 0);
 773         return 0;
 774 }
 775
 776 int eblob_backend_storage_stat(void *priv, struct dnet_stat *st)
 777 {
 778         int err;
 779         struct eblob_backend_config *r = priv;
 780
 781         memset(st, 0, sizeof(struct dnet_stat));
 782
 783         err = backend_stat_low_level(r->data.file, st);
 784         if (err) {
 785                 char root[strlen(r->data.file)+1], *ptr;
 786
 787                 snprintf(root, sizeof(root), "%s", r->data.file);
 788                 ptr = strrchr(root, '/');
 789                 if (ptr) {
 790                         *ptr = '\0';
 791                         err = backend_stat_low_level(root, st);
 792                 }
 793
 794                 if (err)
 795                         return err;
 796         }
 797
 798         return 0;
 799 }
 800
 801 static void eblob_backend_cleanup(void *priv)
 802 {
 803         struct eblob_backend_config *c = priv;
 804
 805         eblob_cleanup(c->eblob);
 806
 807         free(c->data.file);
 808 }
 809
 810 static ssize_t dnet_eblob_db_read(void *priv, struct dnet_raw_id *id, void **datap)
 811 {
 812         struct eblob_backend_config *c = priv;
 813         return dnet_db_read_raw(c->eblob, id, datap);
 814 }
 815
 816 static int dnet_eblob_db_write(void *priv, struct dnet_raw_id *id, void *data, size_t size)
 817 {
 818         struct eblob_backend_config *c = priv;
 819         return dnet_db_write_raw(c->eblob, id, data, size);
 820 }
 821
 822 static int dnet_eblob_db_remove(void *priv, struct dnet_raw_id *id, int real_del)
 823 {
 824         struct eblob_backend_config *c = priv;
 825         return dnet_db_remove_raw(c->eblob, id, real_del);
 826 }
 827
 828 static long long dnet_eblob_db_total_elements(void *priv)
 829 {
 830         struct eblob_backend_config *c = priv;
 831         return eblob_total_elements(c->eblob);
 832 }
 833
 834 static int dnet_eblob_db_iterate(struct dnet_iterate_ctl *ctl)
 835 {
 836         struct eblob_backend_config *c = ctl->iterate_private;
 837         return dnet_db_iterate(c->eblob, ctl);
 838 }
 839
 840 static int dnet_blob_config_init(struct dnet_config_backend *b, struct dnet_config *cfg)
 841 {
 842         struct eblob_backend_config *c = b->data;
 843         int err = 0;
 844
 845         if (!c->data.file) {
 846                 dnet_backend_log(DNET_LOG_ERROR, "blob: no data file present. Exiting.\n");
 847                 err = -EINVAL;
 848                 goto err_out_exit;
 849         }
 850
 851         c->data.log = (struct eblob_log *)b->log;
 852
 853         c->eblob = eblob_init(&c->data);
 854         if (!c->eblob) {
 855                 err = -EINVAL;
 856                 goto err_out_exit;
 857         }
 858
 859         cfg->cb = &b->cb;
 860         cfg->storage_size = b->storage_size;
 861         cfg->storage_free = b->storage_free;
 862         b->cb.storage_stat = eblob_backend_storage_stat;
 863
 864         b->cb.command_private = c;
 865         b->cb.command_handler = eblob_backend_command_handler;
 866         b->cb.send = eblob_send;
 867         b->cb.backend_cleanup = eblob_backend_cleanup;
 868         b->cb.checksum = eblob_backend_checksum;
 869
 870         b->cb.meta_read = dnet_eblob_db_read;
 871         b->cb.meta_write = dnet_eblob_db_write;
 872         b->cb.meta_remove = dnet_eblob_db_remove;
 873         b->cb.meta_total_elements = dnet_eblob_db_total_elements;
 874         b->cb.meta_iterate = dnet_eblob_db_iterate;
 875
 876         return 0;
 877
 878 err_out_exit:
 879         return err;
 880 }
 881
 882 static void dnet_blob_config_cleanup(struct dnet_config_backend *b)
 883 {
 884         struct eblob_backend_config *c = b->data;
 885
 886         eblob_backend_cleanup(c);
 887 }
 888
 889 static struct dnet_config_entry dnet_cfg_entries_blobsystem[] = {
 890         {"sync", dnet_blob_set_sync},
 891         {"data", dnet_blob_set_data},
 892         {"data_block_size", dnet_blob_set_block_size},
 893         {"blob_flags", dnet_blob_set_blob_flags},
 894         {"iterate_thread_num", dnet_blob_set_iterate_thread_num},
 895         {"blob_size", dnet_blob_set_blob_size},
 896         {"records_in_blob", dnet_blob_set_records_in_blob},
 897         {"blob_cache_size", dnet_blob_set_blob_cache_size},
 898         {"defrag_timeout", dnet_blob_set_defrag_timeout},
 899         {"defrag_percentage", dnet_blob_set_defrag_percentage},
 900 };
 901
 902 static struct dnet_config_backend dnet_eblob_backend = {
 903         .name                   = "blob",
 904         .ent                    = dnet_cfg_entries_blobsystem,
 905         .num                    = ARRAY_SIZE(dnet_cfg_entries_blobsystem),
 906         .size                   = sizeof(struct eblob_backend_config),
 907         .init                   = dnet_blob_config_init,
 908         .cleanup                = dnet_blob_config_cleanup,
 909 };
 910
 911 int dnet_eblob_backend_init(void)
 912 {
 913         return dnet_backend_register(&dnet_eblob_backend);
 914 }
 915
 916 void dnet_eblob_backend_exit(void)
 917 {
 918         /* cleanup routing will be called explicitly through backend->cleanup() callback */
 919 }