source4/lib/ldb/modules/paged_results.c

   1 /*
   2    ldb database library
   3
   4    Copyright (C) Simo Sorce  2005-2006
   5
   6      ** NOTE! The following LGPL license applies to the ldb
   7      ** library. This does NOT imply that all of Samba is released
   8      ** under the LGPL
   9
  10    This library is free software; you can redistribute it and/or
  11    modify it under the terms of the GNU Lesser General Public
  12    License as published by the Free Software Foundation; either
  13    version 2 of the License, or (at your option) any later version.
  14
  15    This library is distributed in the hope that it will be useful,
  16    but WITHOUT ANY WARRANTY; without even the implied warranty of
  17    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  18    Lesser General Public License for more details.
  19
  20    You should have received a copy of the GNU Lesser General Public
  21    License along with this library; if not, write to the Free Software
  22    Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
  23 */
  24
  25 /*
  26  *  Name: paged_result
  27  *
  28  *  Component: ldb paged results control module
  29  *
  30  *  Description: this module caches a complete search and sends back
  31  *               results in chunks as asked by the client
  32  *
  33  *  Author: Simo Sorce
  34  */
  35
  36 #include "includes.h"
  37 #include "ldb/include/includes.h"
  38
  39 struct message_store {
  40         /* keep the whole ldb_reply as an optimization
  41          * instead of freeing and talloc-ing the container
  42          * on each result */
  43         struct ldb_reply *r;
  44         struct message_store *next;
  45 };
  46
  47 struct private_data;
  48
  49 struct results_store {
  50
  51         struct private_data *priv;
  52
  53         char *cookie;
  54         time_t timestamp;
  55
  56         struct results_store *prev;
  57         struct results_store *next;
  58
  59         struct message_store *first;
  60         struct message_store *last;
  61         int num_entries;
  62
  63         struct message_store *first_ref;
  64         struct message_store *last_ref;
  65
  66         struct ldb_control **controls;
  67
  68         struct ldb_request *req;
  69 };
  70
  71 struct private_data {
  72
  73         int next_free_id;
  74         struct results_store *store;
  75
  76 };
  77
  78 int store_destructor(struct results_store *store)
  79 {
  80         if (store->prev) {
  81                 store->prev->next = store->next;
  82         }
  83         if (store->next) {
  84                 store->next->prev = store->prev;
  85         }
  86
  87         if (store == store->priv->store) {
  88                 store->priv->store = NULL;
  89         }
  90
  91         return 0;
  92 }
  93
  94 static struct results_store *new_store(struct private_data *priv)
  95 {
  96         struct results_store *newr;
  97         int new_id = priv->next_free_id++;
  98
  99         /* TODO: we should have a limit on the number of
 100          * outstanding paged searches
 101          */
 102
 103         newr = talloc(priv, struct results_store);
 104         if (!newr) return NULL;
 105
 106         newr->priv = priv;
 107
 108         newr->cookie = talloc_asprintf(newr, "%d", new_id);
 109         if (!newr->cookie) {
 110                 talloc_free(newr);
 111                 return NULL;
 112         }
 113
 114         newr->timestamp = time(NULL);
 115
 116         newr->first = NULL;
 117         newr->num_entries = 0;
 118         newr->first_ref = NULL;
 119         newr->controls = NULL;
 120
 121         /* put this entry as first */
 122         newr->prev = NULL;
 123         newr->next = priv->store;
 124         if (priv->store != NULL) priv->store->prev = newr;
 125         priv->store = newr;
 126
 127         talloc_set_destructor(newr, store_destructor);
 128
 129         return newr;
 130 }
 131
 132 struct paged_context {
 133         struct ldb_module *module;
 134         void *up_context;
 135         int (*up_callback)(struct ldb_context *, void *, struct ldb_reply *);
 136
 137         int size;
 138
 139         struct results_store *store;
 140 };
 141
 142 static struct ldb_handle *init_handle(void *mem_ctx, struct ldb_module *module,
 143                                             void *context,
 144                                             int (*callback)(struct ldb_context *, void *, struct ldb_reply *))
 145 {
 146         struct paged_context *ac;
 147         struct ldb_handle *h;
 148
 149         h = talloc_zero(mem_ctx, struct ldb_handle);
 150         if (h == NULL) {
 151                 ldb_set_errstring(module->ldb, "Out of Memory");
 152                 return NULL;
 153         }
 154
 155         h->module = module;
 156
 157         ac = talloc_zero(h, struct paged_context);
 158         if (ac == NULL) {
 159                 ldb_set_errstring(module->ldb, "Out of Memory");
 160                 talloc_free(h);
 161                 return NULL;
 162         }
 163
 164         h->private_data = (void *)ac;
 165
 166         h->state = LDB_ASYNC_INIT;
 167         h->status = LDB_SUCCESS;
 168
 169         ac->module = module;
 170         ac->up_context = context;
 171         ac->up_callback = callback;
 172
 173         return h;
 174 }
 175
 176 static int paged_search_callback(struct ldb_context *ldb, void *context, struct ldb_reply *ares)
 177 {
 178         struct paged_context *ac = NULL;
 179
 180         if (!context || !ares) {
 181                 ldb_set_errstring(ldb, "NULL Context or Result in callback");
 182                 goto error;
 183         }
 184
 185         ac = talloc_get_type(context, struct paged_context);
 186
 187         if (ares->type == LDB_REPLY_ENTRY) {
 188                 if (ac->store->first == NULL) {
 189                         ac->store->first = ac->store->last = talloc(ac->store, struct message_store);
 190                 } else {
 191                         ac->store->last->next = talloc(ac->store, struct message_store);
 192                         ac->store->last = ac->store->last->next;
 193                 }
 194                 if (ac->store->last == NULL) {
 195                         goto error;
 196                 }
 197
 198                 ac->store->num_entries++;
 199
 200                 ac->store->last->r = talloc_steal(ac->store->last, ares);
 201                 ac->store->last->next = NULL;
 202         }
 203
 204         if (ares->type == LDB_REPLY_REFERRAL) {
 205                 if (ac->store->first_ref == NULL) {
 206                         ac->store->first_ref = ac->store->last_ref = talloc(ac->store, struct message_store);
 207                 } else {
 208                         ac->store->last_ref->next = talloc(ac->store, struct message_store);
 209                         ac->store->last_ref = ac->store->last_ref->next;
 210                 }
 211                 if (ac->store->last_ref == NULL) {
 212                         goto error;
 213                 }
 214
 215                 ac->store->last_ref->r = talloc_steal(ac->store->last, ares);
 216                 ac->store->last_ref->next = NULL;
 217         }
 218
 219         if (ares->type == LDB_REPLY_DONE) {
 220                 ac->store->controls = talloc_move(ac->store, &ares->controls);
 221                 talloc_free(ares);
 222         }
 223
 224         return LDB_SUCCESS;
 225
 226 error:
 227         talloc_free(ares);
 228         return LDB_ERR_OPERATIONS_ERROR;
 229 }
 230
 231 static int paged_search(struct ldb_module *module, struct ldb_request *req)
 232 {
 233         struct ldb_control *control;
 234         struct private_data *private_data;
 235         struct ldb_paged_control *paged_ctrl;
 236         struct ldb_control **saved_controls;
 237         struct paged_context *ac;
 238         struct ldb_handle *h;
 239         int ret;
 240
 241         /* check if there's a paged request control */
 242         control = get_control_from_list(req->controls, LDB_CONTROL_PAGED_RESULTS_OID);
 243         if (control == NULL) {
 244                 /* not found go on */
 245                 return ldb_next_request(module, req);
 246         }
 247
 248         private_data = talloc_get_type(module->private_data, struct private_data);
 249
 250         req->handle = NULL;
 251
 252         if (!req->callback || !req->context) {
 253                 ldb_set_errstring(module->ldb,
 254                                   "Async interface called with NULL callback function or NULL context");
 255                 return LDB_ERR_OPERATIONS_ERROR;
 256         }
 257
 258         paged_ctrl = talloc_get_type(control->data, struct ldb_paged_control);
 259         if (!paged_ctrl) {
 260                 return LDB_ERR_PROTOCOL_ERROR;
 261         }
 262
 263         h = init_handle(req, module, req->context, req->callback);
 264         if (!h) {
 265                 return LDB_ERR_OPERATIONS_ERROR;
 266         }
 267         ac = talloc_get_type(h->private_data, struct paged_context);
 268
 269         ac->size = paged_ctrl->size;
 270
 271         /* check if it is a continuation search the store */
 272         if (paged_ctrl->cookie_len == 0) {
 273
 274                 ac->store = new_store(private_data);
 275                 if (ac->store == NULL) {
 276                         talloc_free(h);
 277                         return LDB_ERR_UNWILLING_TO_PERFORM;
 278                 }
 279
 280                 ac->store->req = talloc(ac->store, struct ldb_request);
 281                 if (!ac->store->req)
 282                         return LDB_ERR_OPERATIONS_ERROR;
 283
 284                 ac->store->req->operation = req->operation;
 285                 ac->store->req->op.search.base = req->op.search.base;
 286                 ac->store->req->op.search.scope = req->op.search.scope;
 287                 ac->store->req->op.search.tree = req->op.search.tree;
 288                 ac->store->req->op.search.attrs = req->op.search.attrs;
 289                 ac->store->req->controls = req->controls;
 290
 291                 /* save it locally and remove it from the list */
 292                 /* we do not need to replace them later as we
 293                  * are keeping the original req intact */
 294                 if (!save_controls(control, ac->store->req, &saved_controls)) {
 295                         return LDB_ERR_OPERATIONS_ERROR;
 296                 }
 297
 298                 ac->store->req->context = ac;
 299                 ac->store->req->callback = paged_search_callback;
 300                 ldb_set_timeout_from_prev_req(module->ldb, req, ac->store->req);
 301
 302                 ret = ldb_next_request(module, ac->store->req);
 303
 304         } else {
 305                 struct results_store *current = NULL;
 306
 307                 for (current = private_data->store; current; current = current->next) {
 308                         if (strcmp(current->cookie, paged_ctrl->cookie) == 0) {
 309                                 current->timestamp = time(NULL);
 310                                 break;
 311                         }
 312                 }
 313                 if (current == NULL) {
 314                         talloc_free(h);
 315                         return LDB_ERR_UNWILLING_TO_PERFORM;
 316                 }
 317
 318                 ac->store = current;
 319                 ret = LDB_SUCCESS;
 320         }
 321
 322         req->handle = h;
 323
 324         /* check if it is an abandon */
 325         if (ac->size == 0) {
 326                 talloc_free(ac->store);
 327                 h->status = LDB_SUCCESS;
 328                 h->state = LDB_ASYNC_DONE;
 329                 return LDB_SUCCESS;
 330         }
 331
 332         /* TODO: age out old outstanding requests */
 333
 334         return ret;
 335
 336 }
 337
 338 static int paged_results(struct ldb_handle *handle)
 339 {
 340         struct paged_context *ac;
 341         struct ldb_paged_control *paged;
 342         struct ldb_reply *ares;
 343         struct message_store *msg;
 344         int i, num_ctrls, ret;
 345
 346         ac = talloc_get_type(handle->private_data, struct paged_context);
 347
 348         if (ac->store == NULL)
 349                 return LDB_ERR_OPERATIONS_ERROR;
 350
 351         while (ac->store->num_entries > 0 && ac->size > 0) {
 352                 msg = ac->store->first;
 353                 ret = ac->up_callback(ac->module->ldb, ac->up_context, msg->r);
 354                 if (ret != LDB_SUCCESS) {
 355                         handle->status = ret;
 356                         handle->state = LDB_ASYNC_DONE;
 357                         return ret;
 358                 }
 359
 360                 ac->store->first = msg->next;
 361                 talloc_free(msg);
 362                 ac->store->num_entries--;
 363                 ac->size--;
 364         }
 365
 366         handle->state = LDB_ASYNC_DONE;
 367
 368         while (ac->store->first_ref != NULL) {
 369                 msg = ac->store->first_ref;
 370                 ret = ac->up_callback(ac->module->ldb, ac->up_context, msg->r);
 371                 if (ret != LDB_SUCCESS) {
 372                         handle->status = ret;
 373                         handle->state = LDB_ASYNC_DONE;
 374                         return ret;
 375                 }
 376
 377                 ac->store->first_ref = msg->next;
 378                 talloc_free(msg);
 379         }
 380
 381         ares = talloc_zero(ac->store, struct ldb_reply);
 382         if (ares == NULL) {
 383                 handle->status = LDB_ERR_OPERATIONS_ERROR;
 384                 return handle->status;
 385         }
 386         num_ctrls = 2;
 387         i = 0;
 388
 389         if (ac->store->controls != NULL) {
 390                 ares->controls = ac->store->controls;
 391                 while (ares->controls[i]) i++; /* counting */
 392
 393                 ares->controls = talloc_move(ares, &ac->store->controls);
 394                 num_ctrls += i;
 395         }
 396
 397         ares->controls = talloc_realloc(ares, ares->controls, struct ldb_control *, num_ctrls);
 398         if (ares->controls == NULL) {
 399                 handle->status = LDB_ERR_OPERATIONS_ERROR;
 400                 return handle->status;
 401         }
 402
 403         ares->controls[i] = talloc(ares->controls, struct ldb_control);
 404         if (ares->controls[i] == NULL) {
 405                 handle->status = LDB_ERR_OPERATIONS_ERROR;
 406                 return handle->status;
 407         }
 408
 409         ares->controls[i]->oid = talloc_strdup(ares->controls[i], LDB_CONTROL_PAGED_RESULTS_OID);
 410         if (ares->controls[i]->oid == NULL) {
 411                 handle->status = LDB_ERR_OPERATIONS_ERROR;
 412                 return handle->status;
 413         }
 414
 415         ares->controls[i]->critical = 0;
 416         ares->controls[i + 1] = NULL;
 417
 418         paged = talloc(ares->controls[i], struct ldb_paged_control);
 419         if (paged == NULL) {
 420                 handle->status = LDB_ERR_OPERATIONS_ERROR;
 421                 return handle->status;
 422         }
 423
 424         ares->controls[i]->data = paged;
 425
 426         if (ac->size > 0) {
 427                 paged->size = 0;
 428                 paged->cookie = NULL;
 429                 paged->cookie_len = 0;
 430         } else {
 431                 paged->size = ac->store->num_entries;
 432                 paged->cookie = talloc_strdup(paged, ac->store->cookie);
 433                 paged->cookie_len = strlen(paged->cookie) + 1;
 434         }
 435
 436         ares->type = LDB_REPLY_DONE;
 437
 438         ret = ac->up_callback(ac->module->ldb, ac->up_context, ares);
 439
 440         handle->status = ret;
 441
 442         return ret;
 443 }
 444
 445 static int paged_wait(struct ldb_handle *handle, enum ldb_wait_type type)
 446 {
 447         struct paged_context *ac;
 448         int ret;
 449
 450         if (!handle || !handle->private_data) {
 451                 return LDB_ERR_OPERATIONS_ERROR;
 452         }
 453
 454         if (handle->state == LDB_ASYNC_DONE) {
 455                 return handle->status;
 456         }
 457
 458         handle->state = LDB_ASYNC_PENDING;
 459
 460         ac = talloc_get_type(handle->private_data, struct paged_context);
 461
 462         if (ac->store->req->handle->state == LDB_ASYNC_DONE) {
 463                 /* if lower level is finished we do not need to call it anymore */
 464                 /* return all we have until size == 0 or we empty storage */
 465                 ret = paged_results(handle);
 466
 467                 /* we are done, if num_entries is zero free the storage
 468                  * as that mean we delivered the last batch */
 469                 if (ac->store->num_entries == 0) {
 470                         talloc_free(ac->store);
 471                 }
 472
 473                 return ret;
 474         }
 475
 476         if (type == LDB_WAIT_ALL) {
 477                 while (ac->store->req->handle->state != LDB_ASYNC_DONE) {
 478                         ret = ldb_wait(ac->store->req->handle, type);
 479                         if (ret != LDB_SUCCESS) {
 480                                 handle->state = LDB_ASYNC_DONE;
 481                                 handle->status = ret;
 482                                 return ret;
 483                         }
 484                 }
 485
 486                 ret = paged_results(handle);
 487
 488                 /* we are done, if num_entries is zero free the storage
 489                  * as that mean we delivered the last batch */
 490                 if (ac->store->num_entries == 0) {
 491                         talloc_free(ac->store);
 492                 }
 493
 494                 return ret;
 495         }
 496
 497         ret = ldb_wait(ac->store->req->handle, type);
 498         if (ret != LDB_SUCCESS) {
 499                 handle->state = LDB_ASYNC_DONE;
 500                 handle->status = ret;
 501                 return ret;
 502         }
 503
 504         handle->status = ret;
 505
 506         if (ac->store->num_entries >= ac->size ||
 507             ac->store->req->handle->state == LDB_ASYNC_DONE) {
 508
 509                 ret = paged_results(handle);
 510
 511                 /* we are done, if num_entries is zero free the storage
 512                  * as that mean we delivered the last batch */
 513                 if (ac->store->num_entries == 0) {
 514                         talloc_free(ac->store);
 515                 }
 516         }
 517
 518         return ret;
 519 }
 520
 521 static int paged_request_init(struct ldb_module *module)
 522 {
 523         struct private_data *data;
 524         struct ldb_request *req;
 525         int ret;
 526
 527         data = talloc(module, struct private_data);
 528         if (data == NULL) {
 529                 return LDB_ERR_OTHER;
 530         }
 531
 532         data->next_free_id = 1;
 533         data->store = NULL;
 534         module->private_data = data;
 535
 536         req = talloc(module, struct ldb_request);
 537         if (req == NULL) {
 538                 return LDB_ERR_OPERATIONS_ERROR;
 539         }
 540
 541         req->operation = LDB_REQ_REGISTER_CONTROL;
 542         req->op.reg_control.oid = LDB_CONTROL_PAGED_RESULTS_OID;
 543         req->controls = NULL;
 544
 545         ret = ldb_request(module->ldb, req);
 546         if (ret != LDB_SUCCESS) {
 547                 ldb_debug(module->ldb, LDB_DEBUG_WARNING, "paged_request: Unable to register control with rootdse!\n");
 548         }
 549
 550         talloc_free(req);
 551         return ldb_next_init(module);
 552 }
 553
 554 static const struct ldb_module_ops paged_ops = {
 555         .name           = "paged_results",
 556         .search         = paged_search,
 557         .wait           = paged_wait,
 558         .init_context   = paged_request_init
 559 };
 560
 561 int ldb_paged_results_init(void)
 562 {
 563         return ldb_register_module(&paged_ops);
 564 }
 565