source/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 int store_destructor(struct results_store *store)
  81 {
  82         if (store->prev) {
  83                 store->prev->next = store->next;
  84         }
  85         if (store->next) {
  86                 store->next->prev = store->prev;
  87         }
  88
  89         if (store == store->priv->store) {
  90                 store->priv->store = NULL;
  91         }
  92
  93         return 0;
  94 }
  95
  96 static struct results_store *new_store(struct private_data *priv)
  97 {
  98         struct results_store *newr;
  99         int new_id = priv->next_free_id++;
 100
 101         /* TODO: we should have a limit on the number of
 102          * outstanding paged searches
 103          */
 104
 105         newr = talloc(priv, struct results_store);
 106         if (!newr) return NULL;
 107
 108         newr->priv = priv;
 109
 110         newr->cookie = talloc_asprintf(newr, "%d", new_id);
 111         if (!newr->cookie) {
 112                 talloc_free(newr);
 113                 return NULL;
 114         }
 115
 116         newr->timestamp = time(NULL);
 117
 118         newr->first = NULL;
 119         newr->num_entries = 0;
 120         newr->first_ref = NULL;
 121         newr->controls = NULL;
 122
 123         /* put this entry as first */
 124         newr->prev = NULL;
 125         newr->next = priv->store;
 126         if (priv->store != NULL) priv->store->prev = newr;
 127         priv->store = newr;
 128
 129         talloc_set_destructor(newr, store_destructor);
 130
 131         return newr;
 132 }
 133
 134 struct paged_context {
 135         struct ldb_module *module;
 136         void *up_context;
 137         int (*up_callback)(struct ldb_context *, void *, struct ldb_reply *);
 138
 139         int size;
 140
 141         struct results_store *store;
 142 };
 143
 144 static struct ldb_handle *init_handle(void *mem_ctx, struct ldb_module *module,
 145                                             void *context,
 146                                             int (*callback)(struct ldb_context *, void *, struct ldb_reply *))
 147 {
 148         struct paged_context *ac;
 149         struct ldb_handle *h;
 150
 151         h = talloc_zero(mem_ctx, struct ldb_handle);
 152         if (h == NULL) {
 153                 ldb_set_errstring(module->ldb, "Out of Memory");
 154                 return NULL;
 155         }
 156
 157         h->module = module;
 158
 159         ac = talloc_zero(h, struct paged_context);
 160         if (ac == NULL) {
 161                 ldb_set_errstring(module->ldb, "Out of Memory");
 162                 talloc_free(h);
 163                 return NULL;
 164         }
 165
 166         h->private_data = (void *)ac;
 167
 168         h->state = LDB_ASYNC_INIT;
 169         h->status = LDB_SUCCESS;
 170
 171         ac->module = module;
 172         ac->up_context = context;
 173         ac->up_callback = callback;
 174
 175         return h;
 176 }
 177
 178 static int paged_search_callback(struct ldb_context *ldb, void *context, struct ldb_reply *ares)
 179 {
 180         struct paged_context *ac = NULL;
 181
 182         if (!context || !ares) {
 183                 ldb_set_errstring(ldb, "NULL Context or Result in callback");
 184                 goto error;
 185         }
 186
 187         ac = talloc_get_type(context, struct paged_context);
 188
 189         if (ares->type == LDB_REPLY_ENTRY) {
 190                 if (ac->store->first == NULL) {
 191                         ac->store->first = ac->store->last = talloc(ac->store, struct message_store);
 192                 } else {
 193                         ac->store->last->next = talloc(ac->store, struct message_store);
 194                         ac->store->last = ac->store->last->next;
 195                 }
 196                 if (ac->store->last == NULL) {
 197                         goto error;
 198                 }
 199
 200                 ac->store->num_entries++;
 201
 202                 ac->store->last->r = talloc_steal(ac->store->last, ares);
 203                 ac->store->last->next = NULL;
 204         }
 205
 206         if (ares->type == LDB_REPLY_REFERRAL) {
 207                 if (ac->store->first_ref == NULL) {
 208                         ac->store->first_ref = ac->store->last_ref = talloc(ac->store, struct message_store);
 209                 } else {
 210                         ac->store->last_ref->next = talloc(ac->store, struct message_store);
 211                         ac->store->last_ref = ac->store->last_ref->next;
 212                 }
 213                 if (ac->store->last_ref == NULL) {
 214                         goto error;
 215                 }
 216
 217                 ac->store->last_ref->r = talloc_steal(ac->store->last, ares);
 218                 ac->store->last_ref->next = NULL;
 219         }
 220
 221         if (ares->type == LDB_REPLY_DONE) {
 222                 ac->store->controls = talloc_move(ac->store, &ares->controls);
 223                 talloc_free(ares);
 224         }
 225
 226         return LDB_SUCCESS;
 227
 228 error:
 229         talloc_free(ares);
 230         return LDB_ERR_OPERATIONS_ERROR;
 231 }
 232
 233 static int paged_search(struct ldb_module *module, struct ldb_request *req)
 234 {
 235         struct ldb_control *control;
 236         struct private_data *private_data;
 237         struct ldb_paged_control *paged_ctrl;
 238         struct ldb_control **saved_controls;
 239         struct paged_context *ac;
 240         struct ldb_handle *h;
 241         int ret;
 242
 243         /* check if there's a paged request control */
 244         control = get_control_from_list(req->controls, LDB_CONTROL_PAGED_RESULTS_OID);
 245         if (control == NULL) {
 246                 /* not found go on */
 247                 return ldb_next_request(module, req);
 248         }
 249
 250         private_data = talloc_get_type(module->private_data, struct private_data);
 251
 252         req->handle = NULL;
 253
 254         if (!req->callback || !req->context) {
 255                 ldb_set_errstring(module->ldb,
 256                                   "Async interface called with NULL callback function or NULL context");
 257                 return LDB_ERR_OPERATIONS_ERROR;
 258         }
 259
 260         paged_ctrl = talloc_get_type(control->data, struct ldb_paged_control);
 261         if (!paged_ctrl) {
 262                 return LDB_ERR_PROTOCOL_ERROR;
 263         }
 264
 265         h = init_handle(req, module, req->context, req->callback);
 266         if (!h) {
 267                 return LDB_ERR_OPERATIONS_ERROR;
 268         }
 269         ac = talloc_get_type(h->private_data, struct paged_context);
 270
 271         ac->size = paged_ctrl->size;
 272
 273         /* check if it is a continuation search the store */
 274         if (paged_ctrl->cookie_len == 0) {
 275
 276                 ac->store = new_store(private_data);
 277                 if (ac->store == NULL) {
 278                         talloc_free(h);
 279                         return LDB_ERR_UNWILLING_TO_PERFORM;
 280                 }
 281
 282                 ac->store->req = talloc(ac->store, struct ldb_request);
 283                 if (!ac->store->req)
 284                         return LDB_ERR_OPERATIONS_ERROR;
 285
 286                 ac->store->req->operation = req->operation;
 287                 ac->store->req->op.search.base = req->op.search.base;
 288                 ac->store->req->op.search.scope = req->op.search.scope;
 289                 ac->store->req->op.search.tree = req->op.search.tree;
 290                 ac->store->req->op.search.attrs = req->op.search.attrs;
 291                 ac->store->req->controls = req->controls;
 292
 293                 /* save it locally and remove it from the list */
 294                 /* we do not need to replace them later as we
 295                  * are keeping the original req intact */
 296                 if (!save_controls(control, ac->store->req, &saved_controls)) {
 297                         return LDB_ERR_OPERATIONS_ERROR;
 298                 }
 299
 300                 ac->store->req->context = ac;
 301                 ac->store->req->callback = paged_search_callback;
 302                 ldb_set_timeout_from_prev_req(module->ldb, req, ac->store->req);
 303
 304                 ret = ldb_next_request(module, ac->store->req);
 305
 306         } else {
 307                 struct results_store *current = NULL;
 308
 309                 for (current = private_data->store; current; current = current->next) {
 310                         if (strcmp(current->cookie, paged_ctrl->cookie) == 0) {
 311                                 current->timestamp = time(NULL);
 312                                 break;
 313                         }
 314                 }
 315                 if (current == NULL) {
 316                         talloc_free(h);
 317                         return LDB_ERR_UNWILLING_TO_PERFORM;
 318                 }
 319
 320                 ac->store = current;
 321                 ret = LDB_SUCCESS;
 322         }
 323
 324         req->handle = h;
 325
 326         /* check if it is an abandon */
 327         if (ac->size == 0) {
 328                 talloc_free(ac->store);
 329                 h->status = LDB_SUCCESS;
 330                 h->state = LDB_ASYNC_DONE;
 331                 return LDB_SUCCESS;
 332         }
 333
 334         /* TODO: age out old outstanding requests */
 335
 336         return ret;
 337
 338 }
 339
 340 static int paged_results(struct ldb_handle *handle)
 341 {
 342         struct paged_context *ac;
 343         struct ldb_paged_control *paged;
 344         struct ldb_reply *ares;
 345         struct message_store *msg;
 346         int i, num_ctrls, ret;
 347
 348         ac = talloc_get_type(handle->private_data, struct paged_context);
 349
 350         if (ac->store == NULL)
 351                 return LDB_ERR_OPERATIONS_ERROR;
 352
 353         while (ac->store->num_entries > 0 && ac->size > 0) {
 354                 msg = ac->store->first;
 355                 ret = ac->up_callback(ac->module->ldb, ac->up_context, msg->r);
 356                 if (ret != LDB_SUCCESS) {
 357                         handle->status = ret;
 358                         handle->state = LDB_ASYNC_DONE;
 359                         return ret;
 360                 }
 361
 362                 ac->store->first = msg->next;
 363                 talloc_free(msg);
 364                 ac->store->num_entries--;
 365                 ac->size--;
 366         }
 367
 368         handle->state = LDB_ASYNC_DONE;
 369
 370         while (ac->store->first_ref != NULL) {
 371                 msg = ac->store->first_ref;
 372                 ret = ac->up_callback(ac->module->ldb, ac->up_context, msg->r);
 373                 if (ret != LDB_SUCCESS) {
 374                         handle->status = ret;
 375                         handle->state = LDB_ASYNC_DONE;
 376                         return ret;
 377                 }
 378
 379                 ac->store->first_ref = msg->next;
 380                 talloc_free(msg);
 381         }
 382
 383         ares = talloc_zero(ac->store, struct ldb_reply);
 384         if (ares == NULL) {
 385                 handle->status = LDB_ERR_OPERATIONS_ERROR;
 386                 return handle->status;
 387         }
 388         num_ctrls = 2;
 389         i = 0;
 390
 391         if (ac->store->controls != NULL) {
 392                 ares->controls = ac->store->controls;
 393                 while (ares->controls[i]) i++; /* counting */
 394
 395                 ares->controls = talloc_move(ares, &ac->store->controls);
 396                 num_ctrls += i;
 397         }
 398
 399         ares->controls = talloc_realloc(ares, ares->controls, struct ldb_control *, num_ctrls);
 400         if (ares->controls == NULL) {
 401                 handle->status = LDB_ERR_OPERATIONS_ERROR;
 402                 return handle->status;
 403         }
 404
 405         ares->controls[i] = talloc(ares->controls, struct ldb_control);
 406         if (ares->controls[i] == NULL) {
 407                 handle->status = LDB_ERR_OPERATIONS_ERROR;
 408                 return handle->status;
 409         }
 410
 411         ares->controls[i]->oid = talloc_strdup(ares->controls[i], LDB_CONTROL_PAGED_RESULTS_OID);
 412         if (ares->controls[i]->oid == NULL) {
 413                 handle->status = LDB_ERR_OPERATIONS_ERROR;
 414                 return handle->status;
 415         }
 416
 417         ares->controls[i]->critical = 0;
 418         ares->controls[i + 1] = NULL;
 419
 420         paged = talloc(ares->controls[i], struct ldb_paged_control);
 421         if (paged == NULL) {
 422                 handle->status = LDB_ERR_OPERATIONS_ERROR;
 423                 return handle->status;
 424         }
 425
 426         ares->controls[i]->data = paged;
 427
 428         if (ac->size > 0) {
 429                 paged->size = 0;
 430                 paged->cookie = NULL;
 431                 paged->cookie_len = 0;
 432         } else {
 433                 paged->size = ac->store->num_entries;
 434                 paged->cookie = talloc_strdup(paged, ac->store->cookie);
 435                 paged->cookie_len = strlen(paged->cookie) + 1;
 436         }
 437
 438         ares->type = LDB_REPLY_DONE;
 439
 440         ret = ac->up_callback(ac->module->ldb, ac->up_context, ares);
 441
 442         handle->status = ret;
 443
 444         return ret;
 445 }
 446
 447 static int paged_wait(struct ldb_handle *handle, enum ldb_wait_type type)
 448 {
 449         struct paged_context *ac;
 450         int ret;
 451
 452         if (!handle || !handle->private_data) {
 453                 return LDB_ERR_OPERATIONS_ERROR;
 454         }
 455
 456         if (handle->state == LDB_ASYNC_DONE) {
 457                 return handle->status;
 458         }
 459
 460         handle->state = LDB_ASYNC_PENDING;
 461
 462         ac = talloc_get_type(handle->private_data, struct paged_context);
 463
 464         if (ac->store->req->handle->state == LDB_ASYNC_DONE) {
 465                 /* if lower level is finished we do not need to call it anymore */
 466                 /* return all we have until size == 0 or we empty storage */
 467                 ret = paged_results(handle);
 468
 469                 /* we are done, if num_entries is zero free the storage
 470                  * as that mean we delivered the last batch */
 471                 if (ac->store->num_entries == 0) {
 472                         talloc_free(ac->store);
 473                 }
 474
 475                 return ret;
 476         }
 477
 478         if (type == LDB_WAIT_ALL) {
 479                 while (ac->store->req->handle->state != LDB_ASYNC_DONE) {
 480                         ret = ldb_wait(ac->store->req->handle, type);
 481                         if (ret != LDB_SUCCESS) {
 482                                 handle->state = LDB_ASYNC_DONE;
 483                                 handle->status = ret;
 484                                 return ret;
 485                         }
 486                 }
 487
 488                 ret = paged_results(handle);
 489
 490                 /* we are done, if num_entries is zero free the storage
 491                  * as that mean we delivered the last batch */
 492                 if (ac->store->num_entries == 0) {
 493                         talloc_free(ac->store);
 494                 }
 495
 496                 return ret;
 497         }
 498
 499         ret = ldb_wait(ac->store->req->handle, type);
 500         if (ret != LDB_SUCCESS) {
 501                 handle->state = LDB_ASYNC_DONE;
 502                 handle->status = ret;
 503                 return ret;
 504         }
 505
 506         handle->status = ret;
 507
 508         if (ac->store->num_entries >= ac->size ||
 509             ac->store->req->handle->state == LDB_ASYNC_DONE) {
 510
 511                 ret = paged_results(handle);
 512
 513                 /* we are done, if num_entries is zero free the storage
 514                  * as that mean we delivered the last batch */
 515                 if (ac->store->num_entries == 0) {
 516                         talloc_free(ac->store);
 517                 }
 518         }
 519
 520         return ret;
 521 }
 522
 523 static int paged_request_init(struct ldb_module *module)
 524 {
 525         struct private_data *data;
 526         struct ldb_request *req;
 527         int ret;
 528
 529         data = talloc(module, struct private_data);
 530         if (data == NULL) {
 531                 return LDB_ERR_OTHER;
 532         }
 533
 534         data->next_free_id = 1;
 535         data->store = NULL;
 536         module->private_data = data;
 537
 538         req = talloc(module, struct ldb_request);
 539         if (req == NULL) {
 540                 return LDB_ERR_OPERATIONS_ERROR;
 541         }
 542
 543         req->operation = LDB_REQ_REGISTER_CONTROL;
 544         req->op.reg_control.oid = LDB_CONTROL_PAGED_RESULTS_OID;
 545         req->controls = NULL;
 546
 547         ret = ldb_request(module->ldb, req);
 548         if (ret != LDB_SUCCESS) {
 549                 ldb_debug(module->ldb, LDB_DEBUG_WARNING, "paged_request: Unable to register control with rootdse!\n");
 550         }
 551
 552         talloc_free(req);
 553         return ldb_next_init(module);
 554 }
 555
 556 static const struct ldb_module_ops paged_ops = {
 557         .name           = "paged_results",
 558         .search         = paged_search,
 559         .wait           = paged_wait,
 560         .init_context   = paged_request_init
 561 };
 562
 563 int ldb_paged_results_init(void)
 564 {
 565         return ldb_register_module(&paged_ops);
 566 }
 567