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