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