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                         err = dnet_write_data_wait(n, &ctl, &result);
 513                         if (err < 0) {
 514                                 goto err_out_free;
 515                         }
 516                         free(result);
 517                         err = 0;
 518                 }
 519         }
 520
 521 err_out_free:
 522         if (id->type == -1)
 523                 free(types);
 524 err_out_exit:
 525         return err;
 526 }
 527
 528 static int blob_file_info(struct eblob_backend_config *c, void *state, struct dnet_cmd *cmd)
 529 {
 530         struct eblob_backend *b = c->eblob;
 531         struct eblob_key key;
 532         uint64_t offset, size;
 533         int fd, err;
 534
 535         memcpy(key.id, cmd->id.id, EBLOB_ID_SIZE);
 536         err = eblob_read(b, &key, &fd, &offset, &size, cmd->id.type);
 537         if (err < 0) {
 538                 dnet_backend_log(DNET_LOG_ERROR, "%s: EBLOB: blob-file-info: info-read: %d: %s.\n",
 539                                 dnet_dump_id(&cmd->id), err, strerror(-err));
 540                 goto err_out_exit;
 541         }
 542
 543         if (size == 0) {
 544                 err = -ENOENT;
 545                 dnet_backend_log(DNET_LOG_INFO, "%s: EBLOB: blob-file-info: info-read: ZERO-SIZE-FILE.\n",
 546                                 dnet_dump_id(&cmd->id));
 547                 goto err_out_exit;
 548         }
 549
 550         err = dnet_send_file_info(state, cmd, fd, offset, size);
 551
 552 err_out_exit:
 553         return err;
 554 }
 555
 556 static int blob_bulk_read(struct eblob_backend_config *c, void *state, struct dnet_cmd *cmd, void *data)
 557 {
 558         int err = -1, ret;
 559         struct dnet_io_attr *io = data;
 560         struct dnet_io_attr *ios = io+1;
 561         uint64_t count = 0;
 562         uint64_t i;
 563
 564         dnet_convert_io_attr(io);
 565         count = io->size / sizeof(struct dnet_io_attr);
 566
 567         for (i = 0; i < count; i++) {
 568                 ret = blob_read(c, state, cmd, &ios[i]);
 569                 if (!ret)
 570                         err = 0;
 571                 else if (err == -1)
 572                         err = ret;
 573         }
 574
 575         return err;
 576 }
 577
 578 static int blob_start_defrag(struct eblob_backend_config *c)
 579 {
 580         return eblob_start_defrag(c->eblob);
 581 }
 582
 583 static int eblob_backend_command_handler(void *state, void *priv, struct dnet_cmd *cmd, void *data)
 584 {
 585         int err;
 586         struct eblob_backend_config *c = priv;
 587         char *path, *p;
 588
 589         switch (cmd->cmd) {
 590                 case DNET_CMD_LOOKUP:
 591                         err = blob_file_info(c, state, cmd);
 592                         break;
 593                 case DNET_CMD_WRITE:
 594                         err = blob_write(c, state, cmd, data);
 595                         break;
 596                 case DNET_CMD_READ:
 597                         err = blob_read(c, state, cmd, data);
 598                         break;
 599                 case DNET_CMD_READ_RANGE:
 600                 case DNET_CMD_DEL_RANGE:
 601                         err = blob_read_range(c, state, cmd, data);
 602                         break;
 603                 case DNET_CMD_STAT:
 604                         path = strdup(c->data.file);
 605                         if (!path) {
 606                                 err = -ENOMEM;
 607                                 break;
 608                         }
 609
 610                         p = strrchr(path, '/');
 611                         if (p) {
 612                                 *p = '\0';
 613                         } else {
 614                                 free(path);
 615                                 path = NULL;
 616                         }
 617
 618                         err = backend_stat(state, path, cmd);
 619                         free(path);
 620                         break;
 621                 case DNET_CMD_DEL:
 622                         err = blob_del(c, cmd);
 623                         break;
 624                 case DNET_CMD_BULK_READ:
 625                         err = blob_bulk_read(c, state, cmd, data);
 626                         break;
 627                 case DNET_CMD_DEFRAG:
 628                         err = blob_start_defrag(c);
 629                         break;
 630                 default:
 631                         err = -EINVAL;
 632                         break;
 633         }
 634
 635         return err;
 636 }
 637
 638 static int dnet_blob_set_sync(struct dnet_config_backend *b, char *key __unused, char *value)
 639 {
 640         struct eblob_backend_config *c = b->data;
 641
 642         c->data.sync = atoi(value);
 643         return 0;
 644 }
 645
 646 static int dnet_blob_set_data(struct dnet_config_backend *b, char *key __unused, char *file)
 647 {
 648         struct eblob_backend_config *c = b->data;
 649         int err;
 650
 651         err = backend_storage_size(b, file);
 652         if (err) {
 653                 char root[strlen(file)+1], *ptr;
 654
 655                 snprintf(root, sizeof(root), "%s", file);
 656                 ptr = strrchr(root, '/');
 657                 if (ptr) {
 658                         *ptr = '\0';
 659                         err = backend_storage_size(b, root);
 660                 }
 661
 662                 if (err)
 663                         return err;
 664         }
 665
 666         free(c->data.file);
 667         c->data.file = strdup(file);
 668         if (!c->data.file)
 669                 return -ENOMEM;
 670
 671         return 0;
 672 }
 673
 674 static int dnet_blob_set_block_size(struct dnet_config_backend *b, char *key __unused, char *value)
 675 {
 676         struct eblob_backend_config *c = b->data;
 677
 678         c->data.bsize = strtoul(value, NULL, 0);
 679         return 0;
 680 }
 681
 682 static int dnet_blob_set_blob_size(struct dnet_config_backend *b, char *key __unused, char *value)
 683 {
 684         struct eblob_backend_config *c = b->data;
 685         uint64_t val = strtoul(value, NULL, 0);
 686
 687         if (strchr(value, 'T'))
 688                 val *= 1024*1024*1024*1024ULL;
 689         else if (strchr(value, 'G'))
 690                 val *= 1024*1024*1024ULL;
 691         else if (strchr(value, 'M'))
 692                 val *= 1024*1024;
 693         else if (strchr(value, 'K'))
 694                 val *= 1024;
 695
 696         c->data.blob_size = val;
 697         return 0;
 698 }
 699
 700 static int dnet_blob_set_records_in_blob(struct dnet_config_backend *b, char *key __unused, char *value)
 701 {
 702         struct eblob_backend_config *c = b->data;
 703         uint64_t val = strtoul(value, NULL, 0);
 704
 705         c->data.records_in_blob = val;
 706         return 0;
 707 }
 708
 709 static int dnet_blob_set_blob_cache_size(struct dnet_config_backend *b, char *key __unused, char *value)
 710 {
 711         struct eblob_backend_config *c = b->data;
 712
 713         c->data.cache_size = strtoul(value, NULL, 0);
 714         return 0;
 715 }
 716
 717 static int dnet_blob_set_defrag_timeout(struct dnet_config_backend *b, char *key __unused, char *value)
 718 {
 719         struct eblob_backend_config *c = b->data;
 720
 721         c->data.defrag_timeout = strtoul(value, NULL, 0);
 722         return 0;
 723 }
 724
 725 static int dnet_blob_set_defrag_percentage(struct dnet_config_backend *b, char *key __unused, char *value)
 726 {
 727         struct eblob_backend_config *c = b->data;
 728
 729         c->data.defrag_percentage = strtoul(value, NULL, 0);
 730         return 0;
 731 }
 732
 733 static int dnet_blob_set_iterate_thread_num(struct dnet_config_backend *b, char *key __unused, char *value)
 734 {
 735         struct eblob_backend_config *c = b->data;
 736
 737         c->data.iterate_threads = strtoul(value, NULL, 0);
 738         return 0;
 739 }
 740
 741 static int dnet_blob_set_blob_flags(struct dnet_config_backend *b, char *key __unused, char *value)
 742 {
 743         struct eblob_backend_config *c = b->data;
 744
 745         c->data.blob_flags = strtoul(value, NULL, 0);
 746         return 0;
 747 }
 748
 749 int eblob_backend_storage_stat(void *priv, struct dnet_stat *st)
 750 {
 751         int err;
 752         struct eblob_backend_config *r = priv;
 753
 754         memset(st, 0, sizeof(struct dnet_stat));
 755
 756         err = backend_stat_low_level(r->data.file, st);
 757         if (err) {
 758                 char root[strlen(r->data.file)+1], *ptr;
 759
 760                 snprintf(root, sizeof(root), "%s", r->data.file);
 761                 ptr = strrchr(root, '/');
 762                 if (ptr) {
 763                         *ptr = '\0';
 764                         err = backend_stat_low_level(root, st);
 765                 }
 766
 767                 if (err)
 768                         return err;
 769         }
 770
 771         return 0;
 772 }
 773
 774 static void eblob_backend_cleanup(void *priv)
 775 {
 776         struct eblob_backend_config *c = priv;
 777
 778         eblob_cleanup(c->eblob);
 779
 780         free(c->data.file);
 781 }
 782
 783 static ssize_t dnet_eblob_db_read(void *priv, struct dnet_raw_id *id, void **datap)
 784 {
 785         struct eblob_backend_config *c = priv;
 786         return dnet_db_read_raw(c->eblob, id, datap);
 787 }
 788
 789 static int dnet_eblob_db_write(void *priv, struct dnet_raw_id *id, void *data, size_t size)
 790 {
 791         struct eblob_backend_config *c = priv;
 792         return dnet_db_write_raw(c->eblob, id, data, size);
 793 }
 794
 795 static int dnet_eblob_db_remove(void *priv, struct dnet_raw_id *id, int real_del)
 796 {
 797         struct eblob_backend_config *c = priv;
 798         return dnet_db_remove_raw(c->eblob, id, real_del);
 799 }
 800
 801 static long long dnet_eblob_db_total_elements(void *priv)
 802 {
 803         struct eblob_backend_config *c = priv;
 804         return eblob_total_elements(c->eblob);
 805 }
 806
 807 static int dnet_eblob_db_iterate(struct dnet_iterate_ctl *ctl)
 808 {
 809         struct eblob_backend_config *c = ctl->iterate_private;
 810         return dnet_db_iterate(c->eblob, ctl);
 811 }
 812
 813 static int dnet_blob_config_init(struct dnet_config_backend *b, struct dnet_config *cfg)
 814 {
 815         struct eblob_backend_config *c = b->data;
 816         int err = 0;
 817
 818         if (!c->data.file) {
 819                 dnet_backend_log(DNET_LOG_ERROR, "blob: no data file present. Exiting.\n");
 820                 err = -EINVAL;
 821                 goto err_out_exit;
 822         }
 823
 824         c->data.log = (struct eblob_log *)b->log;
 825
 826         c->eblob = eblob_init(&c->data);
 827         if (!c->eblob) {
 828                 err = -EINVAL;
 829                 goto err_out_exit;
 830         }
 831
 832         cfg->cb = &b->cb;
 833         cfg->storage_size = b->storage_size;
 834         cfg->storage_free = b->storage_free;
 835         b->cb.storage_stat = eblob_backend_storage_stat;
 836
 837         b->cb.command_private = c;
 838         b->cb.command_handler = eblob_backend_command_handler;
 839         b->cb.send = eblob_send;
 840         b->cb.backend_cleanup = eblob_backend_cleanup;
 841
 842         b->cb.meta_read = dnet_eblob_db_read;
 843         b->cb.meta_write = dnet_eblob_db_write;
 844         b->cb.meta_remove = dnet_eblob_db_remove;
 845         b->cb.meta_total_elements = dnet_eblob_db_total_elements;
 846         b->cb.meta_iterate = dnet_eblob_db_iterate;
 847
 848         return 0;
 849
 850 err_out_exit:
 851         return err;
 852 }
 853
 854 static void dnet_blob_config_cleanup(struct dnet_config_backend *b)
 855 {
 856         struct eblob_backend_config *c = b->data;
 857
 858         eblob_backend_cleanup(c);
 859 }
 860
 861 static struct dnet_config_entry dnet_cfg_entries_blobsystem[] = {
 862         {"sync", dnet_blob_set_sync},
 863         {"data", dnet_blob_set_data},
 864         {"data_block_size", dnet_blob_set_block_size},
 865         {"blob_flags", dnet_blob_set_blob_flags},
 866         {"iterate_thread_num", dnet_blob_set_iterate_thread_num},
 867         {"blob_size", dnet_blob_set_blob_size},
 868         {"records_in_blob", dnet_blob_set_records_in_blob},
 869         {"blob_cache_size", dnet_blob_set_blob_cache_size},
 870         {"defrag_timeout", dnet_blob_set_defrag_timeout},
 871         {"defrag_percentage", dnet_blob_set_defrag_percentage},
 872 };
 873
 874 static struct dnet_config_backend dnet_eblob_backend = {
 875         .name                   = "blob",
 876         .ent                    = dnet_cfg_entries_blobsystem,
 877         .num                    = ARRAY_SIZE(dnet_cfg_entries_blobsystem),
 878         .size                   = sizeof(struct eblob_backend_config),
 879         .init                   = dnet_blob_config_init,
 880         .cleanup                = dnet_blob_config_cleanup,
 881 };
 882
 883 int dnet_eblob_backend_init(void)
 884 {
 885         return dnet_backend_register(&dnet_eblob_backend);
 886 }
 887
 888 void dnet_eblob_backend_exit(void)
 889 {
 890         /* cleanup routing will be called explicitly through backend->cleanup() callback */
 891 }