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