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