meta.cpp

   1 /* Copyright (C) 2010 Red Hat, Inc.
   2
   3    This program is free software: you can redistribute it and/or modify
   4    it under the terms of the GNU General Public License as published by
   5    the Free Software Foundation, either version 3 of the License, or
   6    (at your option) any later version.
   7
   8    This program is distributed in the hope that it will be useful,
   9    but WITHOUT ANY WARRANTY; without even the implied warranty of
  10    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  11    GNU General Public License for more details.
  12
  13    You should have received a copy of the GNU General Public License
  14    along with this program.  If not, see <http://www.gnu.org/licenses/>.  */
  15
  16 #include <config.h>
  17
  18 #include <errno.h>
  19 #include <pthread.h>
  20 #include <stdio.h>
  21 #include <sys/time.h>
  22 #include <iostream>
  23 #include "iwh.h"
  24 #include "meta.h"
  25 #include "query.h"
  26
  27 using namespace std;
  28
  29 /* Mongo (rather antisocially) tries to define this itself. */
  30 #if defined(VERSION)
  31 #undef VERSION
  32 #endif
  33
  34 #include <mongo/client/dbclient.h>
  35 using namespace mongo;
  36
  37 /* TBD: parameterize */
  38 #define MAIN_TBL "repo.main"
  39
  40 /*
  41  * Since the client isn't inherently MT-safe, we serialize access to it
  42  * ourselves.  Fortunately, none of our metadata operations should be very
  43  * long-lived; if they are it probably means our connection is FUBAR and other
  44  * threads will be affected anyway.
  45  */
  46
  47 #define SHOW_CONTENTION
  48
  49 pthread_mutex_t         client_lock     = PTHREAD_MUTEX_INITIALIZER;
  50 #if defined(SHOW_CONTENTION)
  51 #define CLIENT_LOCK do {                                        \
  52         if (pthread_mutex_trylock(&client_lock) != 0) {         \
  53                 cout << "contention in " << __func__ << endl;   \
  54                 pthread_mutex_lock(&client_lock);               \
  55         }                                                       \
  56 } while (0)
  57 #else
  58 #define CLIENT_LOCK     pthread_mutex_lock(&client_lock)
  59 #endif
  60 #define CLIENT_UNLOCK   pthread_mutex_unlock(&client_lock)
  61
  62 void
  63 dbl_to_str (double *foo, char *optr)
  64 {
  65         unsigned int i;
  66         unsigned char *iptr = (unsigned char *)foo;
  67
  68         for (i = 0; i < sizeof(*foo); ++i) {
  69                 optr += sprintf(optr,"%02x",*(iptr++));
  70         }
  71 }
  72
  73 class RepoMeta;
  74 class RepoQuery;
  75
  76 class RepoMeta {
  77
  78 public:
  79                 RepoMeta        ();
  80                 ~RepoMeta       ();
  81
  82         DBClientConnection      client;
  83         char                addr[128];
  84
  85         char *  DidPut          (const char *bucket, const char *key,
  86                                  const char *loc, size_t size);
  87         void    GotCopy         (const char *bucket, const char *key,
  88                                  const char *loc);
  89         char *  HasCopy         (const char *bucket, const char *key,
  90                                  const char *loc);
  91         int     SetValue        (const char *bucket, const char *key,
  92                                  const char *mkey, const char * mvalue);
  93         int     GetValue        (const char *bucket, const char *key,
  94                                  const char *mkey, char ** mvalue);
  95         RepoQuery * NewQuery    (const char *bucket, const char *key,
  96                                  const char * expr);
  97         auto_ptr<DBClientCursor> GetCursor (Query &q);
  98         void    Delete          (const char *bucket, const char *key);
  99         size_t  GetSize         (const char *bucket, const char *key);
 100         int     Check           (const char *bucket, const char *key,
 101                                  const char *depot);
 102         void *  GetAttrList     (const char *bucket, const char *key);
 103 };
 104
 105 class RepoQuery {
 106         RepoMeta &              parent;
 107         DBClientCursor *        curs;
 108         value_t *               expr;
 109 public:
 110                 RepoQuery       (const char *, const char *, const char *,
 111                                  RepoMeta &);
 112                 ~RepoQuery      ();
 113         bool    Next            (void);
 114         char    *bucket;
 115         char    *key;
 116         getter_t getter;
 117 };
 118
 119 RepoMeta *it;
 120
 121 RepoMeta::RepoMeta ()
 122 {
 123         if (!verbose) {
 124                 cout.rdbuf(0);
 125                 cout << "bite me" << endl;
 126         }
 127
 128         // TBD: assemble this string properly
 129         sprintf(addr,"%s:%u",db_host,db_port);
 130         try {
 131                 client.connect(addr);
 132         }
 133         catch (ConnectException &ce) {
 134                 cerr << "server down, no metadata access" << endl;
 135         }
 136 }
 137
 138 extern "C" void
 139 meta_init (void)
 140 {
 141         it = new RepoMeta();
 142 }
 143
 144 RepoMeta::~RepoMeta ()
 145 {
 146 }
 147
 148 extern "C" void
 149 meta_fini (void)
 150 {
 151         delete it;
 152 }
 153
 154 auto_ptr<DBClientCursor>
 155 RepoMeta::GetCursor (Query &q)
 156 {
 157         auto_ptr<DBClientCursor> curs;
 158         bool looping = false;
 159
 160         for (;;) {
 161                 if (!client.isFailed()) {
 162                         curs = client.query(MAIN_TBL,q);
 163                         if (curs.get()) {
 164                                 break;
 165                         }
 166                 }
 167                 if (looping) {
 168                         break;
 169                 }
 170                 try {
 171                         client.connect(addr);
 172                 }
 173                 catch (ConnectException &ce) {
 174                         cerr << "reconnection to " << addr << " failed"
 175                              << endl;
 176                 }
 177                 looping = true;
 178         }
 179
 180         return curs;
 181 }
 182
 183 char *
 184 RepoMeta::DidPut (const char *bucket, const char *key, const char *loc,
 185                   size_t size)
 186 {
 187         BSONObjBuilder                  bb;
 188         struct timeval                  now_tv;
 189         double                          now;
 190         auto_ptr<DBClientCursor>        curs;
 191         Query                           q;
 192         char                            now_str[sizeof(now)*2+1];
 193
 194         gettimeofday(&now_tv,NULL);
 195         now = (double)now_tv.tv_sec + (double)now_tv.tv_usec / 1000000.0;
 196         dbl_to_str(&now,now_str);
 197         cout << "now_str = " << now_str << endl;
 198
 199         q = QUERY("_bucket"<<bucket<<"_key"<<key);
 200         curs = GetCursor(q);
 201         if (!curs.get()) {
 202                 cerr << "DidPut failed for " << bucket << "/" << key << endl;
 203                 return NULL;
 204         }
 205         if (curs->more()) {
 206                 /* Nice functionality, but what an ugly syntax! */
 207                 client.update(MAIN_TBL,q,BSON(
 208                         "$set"<<BSON("_loc"<<BSON_ARRAY(loc))
 209                 <<      "$set"<<BSON("_date"<<now)
 210                 <<      "$set"<<BSON("_etag"<<now_str)
 211                 <<      "$set"<<BSON("_size"<<(long long)size)));
 212 #if 0
 213                 client.update(MAIN_TBL,q,
 214                         BSON("$set"<<BSON("_loc"<<BSON_ARRAY(loc))));
 215                 client.update(MAIN_TBL,q,
 216                         BSON("$set"<<BSON("_date"<<now)));
 217                 client.update(MAIN_TBL,q,
 218                         BSON("$set"<<BSON("_etag"<<now_str)));
 219                 client.update(MAIN_TBL,q,
 220                         BSON("$set"<<BSON("_size"<<(long long)size)));
 221 #endif
 222         }
 223         else {
 224                 bb << "_bucket" << bucket << "_key" << key
 225                    << "_loc" << BSON_ARRAY(loc) << "_date" << now
 226                    << "_etag" << now_str << "_size" << (long long)size;
 227                 client.insert(MAIN_TBL,bb.obj());
 228         }
 229
 230         return strdup(now_str);
 231 }
 232
 233 extern "C" char *
 234 meta_did_put (const char *bucket, const char *key, const char *loc, size_t size)
 235 {
 236         char    *rc;
 237
 238         cout << "meta_did_put(" << bucket << "," << key << "," << loc << ")"
 239              << endl;
 240
 241         CLIENT_LOCK;
 242         rc = it->DidPut(bucket,key,loc,size);
 243         CLIENT_UNLOCK;
 244
 245         return rc;
 246 }
 247
 248 void
 249 RepoMeta::GotCopy (const char *bucket, const char *key, const char *loc)
 250 {
 251         BSONObjBuilder                  bb;
 252         auto_ptr<DBClientCursor>        curs;
 253         Query                           q;
 254
 255         q = QUERY("_bucket"<<bucket<<"_key"<<key);
 256         curs = GetCursor(q);
 257         if (!curs.get()) {
 258                 cerr << "GotCopy failed for " << bucket << "/" << key << endl;
 259                 return;
 260         }
 261         if (curs->more()) {
 262                 /* Nice functionality, but what an ugly syntax! */
 263                 client.update(MAIN_TBL,q,BSON("$addToSet"<<BSON("_loc"<<loc)));
 264         }
 265         else {
 266                 cerr << bucket << "/" << key << " not found in GotCopy!" << endl;
 267         }
 268 }
 269
 270 extern "C" void
 271 meta_got_copy (const char *bucket, const char *key, const char *loc)
 272 {
 273         CLIENT_LOCK;
 274         it->GotCopy(bucket,key,loc);
 275         CLIENT_UNLOCK;
 276 }
 277
 278 char *
 279 RepoMeta::HasCopy (const char *bucket, const char *key, const char *loc)
 280 {
 281         BSONObjBuilder                  bb;
 282         auto_ptr<DBClientCursor>        curs;
 283         Query                           q;
 284         const char                      *value;
 285
 286         q = QUERY("_bucket"<<bucket<<"_key"<<key<<"_loc"<<loc);
 287         curs = GetCursor(q);
 288         if (!curs.get()) {
 289                 cerr << "HasCopy failed for " << bucket << "/" << key << endl;
 290                 return NULL;
 291         }
 292         if (!curs->more()) {
 293                 cout << bucket << "/" << key << " not found at " << loc << endl;
 294                 return (char *)"";
 295         }
 296
 297         value = curs->next().getStringField("_etag");
 298         if (!value || !*value) {
 299                 cout << bucket << "/" << key << " no _etag at " << loc << endl;
 300                 return (char *)"";
 301         }
 302
 303         cout << bucket << "/" << key << " _etag = " << value << endl;
 304         return strdup(value);
 305 }
 306
 307 extern "C" char *
 308 meta_has_copy (const char *bucket, const char *key, const char *loc)
 309 {
 310         char    *rc;
 311
 312         CLIENT_LOCK;
 313         rc = it->HasCopy(bucket,key,loc);
 314         CLIENT_UNLOCK;
 315
 316         return rc;
 317 }
 318
 319 int
 320 RepoMeta::SetValue (const char *bucket, const char *key, const char *mkey,
 321                     const char * mvalue)
 322 {
 323         Query   q       = QUERY("_bucket"<<bucket<<"_key"<<key);
 324
 325         try {
 326                 client.update(MAIN_TBL,q,BSON("$set"<<BSON(mkey<<mvalue)),1);
 327         }
 328         catch (ConnectException &ce) {
 329                 cerr << "SetValue failed for " << bucket << "/" << key << ":"
 330                      << mkey << endl;
 331                 return ENOTCONN;
 332         }
 333
 334         // TBD: check for and propagate errors.
 335         return 0;
 336 }
 337
 338 extern "C" int
 339 meta_set_value (const char *bucket, const char *key, const char *mkey,
 340                 const char * mvalue)
 341 {
 342         int     rc;
 343
 344         CLIENT_LOCK;
 345         rc = it->SetValue(bucket,key,mkey,mvalue);
 346         CLIENT_UNLOCK;
 347
 348         return rc;
 349 }
 350
 351 int
 352 RepoMeta::GetValue (const char *bucket, const char *key, const char *mkey,
 353                     char ** mvalue)
 354 {
 355         auto_ptr<DBClientCursor>        curs;
 356         Query                           q;
 357         BSONObj                         bo;
 358         const char *                    data;
 359
 360         q = QUERY("_bucket"<<bucket<<"_key"<<key);
 361         curs = GetCursor(q);
 362         if (!curs.get()) {
 363                 cerr << "GetValue failed for " << bucket << "/" << key << ":"
 364                      << mkey << endl;
 365                 return ENOTCONN;
 366         }
 367         if (!curs->more()) {
 368                 return ENXIO;
 369         }
 370
 371         bo = curs->next();
 372         data = bo.getStringField(mkey);
 373         if (!data || !*data) {
 374                 return ENXIO;
 375         }
 376
 377         *mvalue = strdup(data);
 378         return 0;
 379 }
 380
 381 extern "C" int
 382 meta_get_value (const char *bucket, const char *key, const char *mkey,
 383                 char ** mvalue)
 384 {
 385         int     rc;
 386
 387         CLIENT_LOCK;
 388         rc = it->GetValue(bucket,key,mkey,mvalue);
 389         CLIENT_UNLOCK;
 390
 391         return rc;
 392 }
 393
 394 RepoQuery::RepoQuery (const char *bucket, const char *key, const char *qstr,
 395                       RepoMeta &p)
 396         : parent(p)
 397 {
 398         Query                           q;
 399         auto_ptr<DBClientCursor>        tmp;
 400
 401         if (bucket) {
 402                 cout << "bucket is " << bucket << " and we don't care" << endl;
 403                 q = QUERY("_bucket"<<bucket);
 404         }
 405         else if (key) {
 406                 cout << "key is " << key << " and we don't care" << endl;
 407                 q = QUERY("_key"<<key);
 408         }
 409         else {
 410                 abort();
 411         }
 412
 413         /*
 414          * TBD: we should really convert our query into one of Mongo's,
 415          * and let them do all the work.  Handling the general case
 416          * would be pretty messy, but we could handle specific cases
 417          * pretty easily.  For example, a very high percentage of
 418          * queries are likely to be a single field/value comparison.
 419          * For now just punt, but revisit later.
 420          */
 421
 422         if (qstr) {
 423                 expr = parse(qstr);
 424                 if (expr) {
 425                         print_value(expr);
 426                 }
 427                 else {
 428                         cout << "could not parse " << qstr << endl;
 429                 }
 430         }
 431         else {
 432                 expr = NULL;
 433         }
 434
 435         curs = parent.GetCursor(q).release();
 436         bucket = NULL;
 437         key = NULL;
 438 }
 439
 440 RepoQuery::~RepoQuery ()
 441 {
 442         cout << "in " << __func__ << endl;
 443         if (expr) {
 444                 free_value(expr);
 445         }
 446
 447         delete curs;
 448 }
 449
 450 extern "C" void
 451 meta_query_stop (void * qobj)
 452 {
 453         CLIENT_LOCK;
 454         delete (RepoQuery *)qobj;
 455         CLIENT_UNLOCK;
 456 }
 457
 458 extern "C" const char *
 459 query_getter (void *ctx, const char *id)
 460 {
 461         BSONObj *cur_bo = (BSONObj *)ctx;
 462
 463         return (char *)cur_bo->getStringField(id);
 464 }
 465
 466 bool
 467 RepoQuery::Next (void)
 468 {
 469         BSONObj         bo;
 470
 471         if (!curs) {
 472                 return false;
 473         }
 474
 475         while (curs->more()) {
 476                 bo = curs->next();
 477                 if (expr) {
 478                         getter.func = query_getter;
 479                         getter.ctx = (void *)&bo;
 480                         if (eval(expr,&getter,NULL) <= 0) {
 481                                 continue;
 482                         }
 483                 }
 484                 bucket = (char *)bo.getStringField("_bucket");
 485                 key = (char *)bo.getStringField("_key");
 486                 return true;
 487         }
 488
 489         return false;
 490 }
 491
 492 RepoQuery *
 493 RepoMeta::NewQuery (const char *bucket, const char *key, const char *expr)
 494 {
 495         return new RepoQuery(bucket,key,expr,*this);
 496 }
 497
 498 extern "C" void *
 499 meta_query_new (const char *bucket, const char *key, const char *expr)
 500 {
 501         void    *rc;
 502
 503         if ((bucket && key) || (!bucket && !key)) {
 504                 return NULL;
 505         }
 506
 507         CLIENT_LOCK;
 508         rc = it->NewQuery(bucket,key,expr);
 509         CLIENT_UNLOCK;
 510
 511         return rc;
 512 }
 513
 514 extern "C" int
 515 meta_query_next (void * qobj, char ** bucket, char ** key)
 516 {
 517         RepoQuery *     rq      = (RepoQuery *)qobj;
 518
 519         CLIENT_LOCK;
 520         if (!rq->Next()) {
 521                 CLIENT_UNLOCK;
 522                 return 0;
 523         }
 524         CLIENT_UNLOCK;
 525
 526         *bucket = rq->bucket;
 527         *key = rq->key;
 528         return 1;
 529 }
 530
 531 #if 0
 532 char *
 533 RepoMeta::BucketList (void)
 534 {
 535         /*
 536          * TBD: make this return values instead of producing output.
 537          * This is just a code fragment showing how to get a list of buckets,
 538          * in case I forget.
 539          */
 540         BSONObj                         repl;
 541
 542         BSONObj dist = BSON("distinct"<<"main"<<"_key"<<"_bucket");
 543         if (client.runCommand("repo",dist,repl)) {
 544                 cout << repl.toString() << endl;
 545                 BSONObj elem = repl.getField("values").embeddedObject();
 546                 for (int i = 0; i < elem.nFields(); ++i) {
 547                         cout << elem[i].str() << endl;
 548                 }
 549         }
 550 }
 551 #endif
 552
 553 void
 554 RepoMeta::Delete (const char *bucket, const char *key)
 555 {
 556         Query   q       = QUERY("_bucket"<<bucket<<"_key"<<key);
 557
 558         try {
 559                 client.remove(MAIN_TBL,q);
 560         }
 561         catch (ConnectException &ce) {
 562                 cerr << "Delete failed for " << bucket << "/" << key << endl;
 563         }
 564 }
 565
 566 extern "C"
 567 void
 568 meta_delete (const char *bucket, const char *key)
 569 {
 570         CLIENT_LOCK;
 571         it->Delete(bucket,key);
 572         CLIENT_UNLOCK;
 573 }
 574
 575 size_t
 576 RepoMeta::GetSize (const char *bucket, const char *key)
 577 {
 578         auto_ptr<DBClientCursor>        curs;
 579         Query                           q;
 580         BSONObj                         bo;
 581         const char *                    data;
 582
 583         (void)data;
 584
 585         q = QUERY("_bucket"<<bucket<<"_key"<<key);
 586         curs = GetCursor(q);
 587
 588         if (!curs->more()) {
 589                 return 0;
 590         }
 591
 592         bo = curs->next();
 593         return bo.getField("_size").numberLong();
 594 }
 595
 596 extern "C"
 597 size_t
 598 meta_get_size (const char *bucket, const char *key)
 599 {
 600         size_t  rc;
 601
 602         CLIENT_LOCK;
 603         rc = it->GetSize(bucket,key);
 604         CLIENT_UNLOCK;
 605
 606         return rc;
 607 }
 608
 609 class AttrList {
 610 public:
 611                                 AttrList        (BSONObj &);
 612         int                     Next            (const char **, const char **);
 613         BSONObj                 obj;
 614         vector<BSONElement>     vec;
 615         int                     idx;
 616 };
 617
 618 AttrList::AttrList (BSONObj &bo)
 619 {
 620         obj = bo.copy();
 621         obj.elems(vec);
 622         idx = 0;
 623 }
 624
 625 int
 626 AttrList::Next (const char **name, const char **value)
 627 {
 628         BSONElement     elem;
 629
 630         while (idx < vec.size()) {
 631                 elem = vec[idx++];
 632                 if (elem.type() == String) {
 633                         *name = elem.fieldName();
 634                         *value = elem.String().c_str();
 635                         return 1;
 636                 }
 637         }
 638
 639         return 0;
 640 }
 641
 642 void *
 643 RepoMeta::GetAttrList (const char *bucket, const char *key)
 644 {
 645         auto_ptr<DBClientCursor>        curs;
 646         Query                           q;
 647         BSONObj                         bo;
 648
 649         q = QUERY("_bucket"<<bucket<<"_key"<<key);
 650         curs = GetCursor(q);
 651
 652         if (!curs->more()) {
 653                 return NULL;
 654         }
 655         bo = curs->next();
 656
 657         return new AttrList(bo);
 658 }
 659 extern "C"
 660 void *
 661 meta_get_attrs (const char *bucket, const char *key)
 662 {
 663         void *poc;
 664
 665         CLIENT_LOCK;
 666         poc = it->GetAttrList(bucket,key);
 667         CLIENT_UNLOCK;
 668
 669         return poc;
 670 }
 671
 672 extern "C"
 673 int
 674 meta_attr_next (void *ctx, const char **name, const char **value)
 675 {
 676         AttrList *poc = (AttrList *)ctx;
 677
 678         return poc->Next(name,value);
 679 }
 680
 681 extern "C"
 682 void
 683 meta_attr_stop (void *ctx)
 684 {
 685         AttrList *poc = (AttrList *)ctx;
 686
 687         delete poc;
 688 }