lib/ldb/modules/paged_results.c

   1 /*
   2    ldb database library
   3
   4    Copyright (C) Simo Sorce  2005-2008
   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 "replace.h"
  36 #include "system/filesys.h"
  37 #include "system/time.h"
  38 #include "dlinklist.h"
  39 #include <assert.h>
  40 #include "ldb_module.h"
  41
  42 struct message_store {
  43         /* keep the whole ldb_reply as an optimization
  44          * instead of freeing and talloc-ing the container
  45          * on each result */
  46         struct ldb_reply *r;
  47         struct message_store *next;
  48 };
  49
  50 struct private_data;
  51
  52 struct results_store {
  53         struct results_store *prev, *next;
  54
  55         struct private_data *priv;
  56
  57         char *cookie;
  58         time_t timestamp;
  59
  60         struct message_store *first;
  61         struct message_store *last;
  62         int num_entries;
  63
  64         struct message_store *first_ref;
  65         struct message_store *last_ref;
  66
  67         struct ldb_control **controls;
  68 };
  69
  70 struct private_data {
  71         uint32_t next_free_id;
  72         size_t num_stores;
  73         struct results_store *store;
  74
  75 };
  76
  77 static int store_destructor(struct results_store *del)
  78 {
  79         struct private_data *priv = del->priv;
  80         DLIST_REMOVE(priv->store, del);
  81
  82         assert(priv->num_stores > 0);
  83         priv->num_stores -= 1;
  84
  85         return 0;
  86 }
  87
  88 static struct results_store *new_store(struct private_data *priv)
  89 {
  90         struct results_store *newr;
  91         uint32_t new_id = priv->next_free_id++;
  92
  93         /* TODO: we should have a limit on the number of
  94          * outstanding paged searches
  95          */
  96
  97         newr = talloc(priv, struct results_store);
  98         if (!newr) return NULL;
  99
 100         newr->priv = priv;
 101
 102         newr->cookie = talloc_asprintf(newr, "%d", new_id);
 103         if (!newr->cookie) {
 104                 talloc_free(newr);
 105                 return NULL;
 106         }
 107
 108         newr->timestamp = time(NULL);
 109
 110         newr->first = NULL;
 111         newr->num_entries = 0;
 112         newr->first_ref = NULL;
 113         newr->controls = NULL;
 114
 115         DLIST_ADD(priv->store, newr);
 116
 117         assert(priv->num_stores < SIZE_MAX);
 118         priv->num_stores += 1;
 119
 120         talloc_set_destructor(newr, store_destructor);
 121
 122         if (priv->num_stores > 10) {
 123                 struct results_store *last;
 124                 /*
 125                  * 10 is the default for MaxResultSetsPerConn --
 126                  * possibly need to parameterize it.
 127                  */
 128                 last = DLIST_TAIL(priv->store);
 129                 TALLOC_FREE(last);
 130         }
 131
 132         return newr;
 133 }
 134
 135 struct paged_context {
 136         struct ldb_module *module;
 137         struct ldb_request *req;
 138
 139         struct results_store *store;
 140         int size;
 141         struct ldb_control **controls;
 142 };
 143
 144 static int paged_results(struct paged_context *ac)
 145 {
 146         struct ldb_paged_control *paged;
 147         struct message_store *msg;
 148         unsigned int i, num_ctrls;
 149         int ret;
 150
 151         if (ac->store == NULL) {
 152                 return LDB_ERR_OPERATIONS_ERROR;
 153         }
 154
 155         while (ac->store->num_entries > 0 && ac->size > 0) {
 156                 msg = ac->store->first;
 157                 ret = ldb_module_send_entry(ac->req, msg->r->message, msg->r->controls);
 158                 if (ret != LDB_SUCCESS) {
 159                         return ret;
 160                 }
 161
 162                 ac->store->first = msg->next;
 163                 talloc_free(msg);
 164                 ac->store->num_entries--;
 165                 ac->size--;
 166         }
 167
 168         while (ac->store->first_ref != NULL) {
 169                 msg = ac->store->first_ref;
 170                 ret = ldb_module_send_referral(ac->req, msg->r->referral);
 171                 if (ret != LDB_SUCCESS) {
 172                         return ret;
 173                 }
 174
 175                 ac->store->first_ref = msg->next;
 176                 talloc_free(msg);
 177         }
 178
 179         /* return result done */
 180         num_ctrls = 1;
 181         i = 0;
 182
 183         if (ac->store->controls != NULL) {
 184                 while (ac->store->controls[i]) i++; /* counting */
 185
 186                 num_ctrls += i;
 187         }
 188
 189         ac->controls = talloc_array(ac, struct ldb_control *, num_ctrls +1);
 190         if (ac->controls == NULL) {
 191                 return LDB_ERR_OPERATIONS_ERROR;
 192         }
 193         ac->controls[num_ctrls] = NULL;
 194
 195         for (i = 0; i < (num_ctrls -1); i++) {
 196                 ac->controls[i] = talloc_reference(ac->controls, ac->store->controls[i]);
 197         }
 198
 199         ac->controls[i] = talloc(ac->controls, struct ldb_control);
 200         if (ac->controls[i] == NULL) {
 201                 return LDB_ERR_OPERATIONS_ERROR;
 202         }
 203
 204         ac->controls[i]->oid = talloc_strdup(ac->controls[i],
 205                                                 LDB_CONTROL_PAGED_RESULTS_OID);
 206         if (ac->controls[i]->oid == NULL) {
 207                 return LDB_ERR_OPERATIONS_ERROR;
 208         }
 209
 210         ac->controls[i]->critical = 0;
 211
 212         paged = talloc(ac->controls[i], struct ldb_paged_control);
 213         if (paged == NULL) {
 214                 return LDB_ERR_OPERATIONS_ERROR;
 215         }
 216
 217         ac->controls[i]->data = paged;
 218
 219         if (ac->size > 0) {
 220                 paged->size = 0;
 221                 paged->cookie = NULL;
 222                 paged->cookie_len = 0;
 223         } else {
 224                 paged->size = ac->store->num_entries;
 225                 paged->cookie = talloc_strdup(paged, ac->store->cookie);
 226                 paged->cookie_len = strlen(paged->cookie) + 1;
 227         }
 228
 229         return LDB_SUCCESS;
 230 }
 231
 232 static int paged_search_callback(struct ldb_request *req, struct ldb_reply *ares)
 233 {
 234         struct paged_context *ac ;
 235         struct message_store *msg_store;
 236         int ret;
 237
 238         ac = talloc_get_type(req->context, struct paged_context);
 239
 240         if (!ares) {
 241                 return ldb_module_done(ac->req, NULL, NULL,
 242                                         LDB_ERR_OPERATIONS_ERROR);
 243         }
 244         if (ares->error != LDB_SUCCESS) {
 245                 return ldb_module_done(ac->req, ares->controls,
 246                                         ares->response, ares->error);
 247         }
 248
 249         switch (ares->type) {
 250         case LDB_REPLY_ENTRY:
 251                 msg_store = talloc(ac->store, struct message_store);
 252                 if (msg_store == NULL) {
 253                         return ldb_module_done(ac->req, NULL, NULL,
 254                                                 LDB_ERR_OPERATIONS_ERROR);
 255                 }
 256                 msg_store->next = NULL;
 257                 msg_store->r = talloc_steal(msg_store, ares);
 258
 259                 if (ac->store->first == NULL) {
 260                         ac->store->first = msg_store;
 261                 } else {
 262                         ac->store->last->next = msg_store;
 263                 }
 264                 ac->store->last = msg_store;
 265
 266                 ac->store->num_entries++;
 267
 268                 break;
 269
 270         case LDB_REPLY_REFERRAL:
 271                 msg_store = talloc(ac->store, struct message_store);
 272                 if (msg_store == NULL) {
 273                         return ldb_module_done(ac->req, NULL, NULL,
 274                                                 LDB_ERR_OPERATIONS_ERROR);
 275                 }
 276                 msg_store->next = NULL;
 277                 msg_store->r = talloc_steal(msg_store, ares);
 278
 279                 if (ac->store->first_ref == NULL) {
 280                         ac->store->first_ref = msg_store;
 281                 } else {
 282                         ac->store->last_ref->next = msg_store;
 283                 }
 284                 ac->store->last_ref = msg_store;
 285
 286                 break;
 287
 288         case LDB_REPLY_DONE:
 289                 ac->store->controls = talloc_move(ac->store, &ares->controls);
 290                 ret = paged_results(ac);
 291                 return ldb_module_done(ac->req, ac->controls,
 292                                         ares->response, ret);
 293         }
 294
 295         return LDB_SUCCESS;
 296 }
 297
 298 static int paged_search(struct ldb_module *module, struct ldb_request *req)
 299 {
 300         struct ldb_context *ldb;
 301         struct ldb_control *control;
 302         struct private_data *private_data;
 303         struct ldb_paged_control *paged_ctrl;
 304         struct ldb_control **saved_controls;
 305         struct ldb_request *search_req;
 306         struct paged_context *ac;
 307         int ret;
 308
 309         ldb = ldb_module_get_ctx(module);
 310
 311         /* check if there's a paged request control */
 312         control = ldb_request_get_control(req, LDB_CONTROL_PAGED_RESULTS_OID);
 313         if (control == NULL) {
 314                 /* not found go on */
 315                 return ldb_next_request(module, req);
 316         }
 317
 318         paged_ctrl = talloc_get_type(control->data, struct ldb_paged_control);
 319         if (!paged_ctrl) {
 320                 return LDB_ERR_PROTOCOL_ERROR;
 321         }
 322
 323         private_data = talloc_get_type(ldb_module_get_private(module),
 324                                         struct private_data);
 325
 326         ac = talloc_zero(req, struct paged_context);
 327         if (ac == NULL) {
 328                 ldb_set_errstring(ldb, "Out of Memory");
 329                 return LDB_ERR_OPERATIONS_ERROR;
 330         }
 331
 332         ac->module = module;
 333         ac->req = req;
 334         ac->size = paged_ctrl->size;
 335         if (ac->size < 0) {
 336                 /* apparently some clients send more than 2^31. This
 337                    violates the ldap standard, but we need to cope */
 338                 ac->size = 0x7FFFFFFF;
 339         }
 340
 341         /* check if it is a continuation search the store */
 342         if (paged_ctrl->cookie_len == 0) {
 343                 if (paged_ctrl->size == 0) {
 344                         return LDB_ERR_OPERATIONS_ERROR;
 345                 }
 346
 347                 ac->store = new_store(private_data);
 348                 if (ac->store == NULL) {
 349                         return LDB_ERR_OPERATIONS_ERROR;
 350                 }
 351
 352                 ret = ldb_build_search_req_ex(&search_req, ldb, ac,
 353                                                 req->op.search.base,
 354                                                 req->op.search.scope,
 355                                                 req->op.search.tree,
 356                                                 req->op.search.attrs,
 357                                                 req->controls,
 358                                                 ac,
 359                                                 paged_search_callback,
 360                                                 req);
 361                 if (ret != LDB_SUCCESS) {
 362                         return ret;
 363                 }
 364
 365                 /* save it locally and remove it from the list */
 366                 /* we do not need to replace them later as we
 367                  * are keeping the original req intact */
 368                 if (!ldb_save_controls(control, search_req, &saved_controls)) {
 369                         return LDB_ERR_OPERATIONS_ERROR;
 370                 }
 371
 372                 return ldb_next_request(module, search_req);
 373
 374         } else {
 375                 struct results_store *current = NULL;
 376
 377                 /* TODO: age out old outstanding requests */
 378                 for (current = private_data->store; current; current = current->next) {
 379                         if (strcmp(current->cookie, paged_ctrl->cookie) == 0) {
 380                                 current->timestamp = time(NULL);
 381                                 break;
 382                         }
 383                 }
 384                 if (current == NULL) {
 385                         return LDB_ERR_UNWILLING_TO_PERFORM;
 386                 }
 387
 388                 DLIST_PROMOTE(private_data->store, current);
 389
 390                 ac->store = current;
 391
 392                 /* check if it is an abandon */
 393                 if (ac->size == 0) {
 394                         return ldb_module_done(req, NULL, NULL,
 395                                                                 LDB_SUCCESS);
 396                 }
 397
 398                 ret = paged_results(ac);
 399                 if (ret != LDB_SUCCESS) {
 400                         return ldb_module_done(req, NULL, NULL, ret);
 401                 }
 402                 return ldb_module_done(req, ac->controls, NULL,
 403                                                                 LDB_SUCCESS);
 404         }
 405 }
 406
 407 static int paged_request_init(struct ldb_module *module)
 408 {
 409         struct ldb_context *ldb;
 410         struct private_data *data;
 411         int ret;
 412
 413         ldb = ldb_module_get_ctx(module);
 414
 415         data = talloc(module, struct private_data);
 416         if (data == NULL) {
 417                 return LDB_ERR_OTHER;
 418         }
 419
 420         data->next_free_id = 1;
 421         data->num_stores = 0;
 422         data->store = NULL;
 423         ldb_module_set_private(module, data);
 424
 425         ret = ldb_mod_register_control(module, LDB_CONTROL_PAGED_RESULTS_OID);
 426         if (ret != LDB_SUCCESS) {
 427                 ldb_debug(ldb, LDB_DEBUG_WARNING,
 428                         "paged_results:"
 429                         "Unable to register control with rootdse!");
 430         }
 431
 432         return ldb_next_init(module);
 433 }
 434
 435 static const struct ldb_module_ops ldb_paged_results_module_ops = {
 436         .name           = "paged_results",
 437         .search         = paged_search,
 438         .init_context   = paged_request_init
 439 };
 440
 441 int ldb_paged_results_init(const char *version)
 442 {
 443         LDB_MODULE_CHECK_VERSION(version);
 444         return ldb_register_module(&ldb_paged_results_module_ops);
 445 }