meta.cpp

   1 /* Copyright (C) 2010-2011 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 static 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
 444         delete curs;
 445 }
 446
 447 extern "C" void
 448 meta_query_stop (void * qobj)
 449 {
 450         CLIENT_LOCK;
 451         delete (RepoQuery *)qobj;
 452         CLIENT_UNLOCK;
 453 }
 454
 455 extern "C" const char *
 456 query_getter (void *ctx, const char *id)
 457 {
 458         BSONObj *cur_bo = (BSONObj *)ctx;
 459
 460         return (char *)cur_bo->getStringField(id);
 461 }
 462
 463 bool
 464 RepoQuery::Next (void)
 465 {
 466         BSONObj         bo;
 467
 468         if (!curs) {
 469                 return false;
 470         }
 471
 472         while (curs->more()) {
 473                 bo = curs->next();
 474                 if (expr) {
 475                         getter.func = query_getter;
 476                         getter.ctx = (void *)&bo;
 477                         if (eval(expr,&getter,NULL) <= 0) {
 478                                 continue;
 479                         }
 480                 }
 481                 bucket = (char *)bo.getStringField("_bucket");
 482                 key = (char *)bo.getStringField("_key");
 483                 return true;
 484         }
 485
 486         return false;
 487 }
 488
 489 RepoQuery *
 490 RepoMeta::NewQuery (const char *bucket, const char *key, const char *expr)
 491 {
 492         return new RepoQuery(bucket,key,expr,*this);
 493 }
 494
 495 extern "C" void *
 496 meta_query_new (const char *bucket, const char *key, const char *expr)
 497 {
 498         void    *rc;
 499
 500         if ((bucket && key) || (!bucket && !key)) {
 501                 return NULL;
 502         }
 503
 504         CLIENT_LOCK;
 505         rc = it->NewQuery(bucket,key,expr);
 506         CLIENT_UNLOCK;
 507
 508         return rc;
 509 }
 510
 511 extern "C" int
 512 meta_query_next (void * qobj, char ** bucket, char ** key)
 513 {
 514         RepoQuery *     rq      = (RepoQuery *)qobj;
 515
 516         CLIENT_LOCK;
 517         if (!rq->Next()) {
 518                 CLIENT_UNLOCK;
 519                 return 0;
 520         }
 521         CLIENT_UNLOCK;
 522
 523         *bucket = rq->bucket;
 524         *key = rq->key;
 525         return 1;
 526 }
 527
 528 #if 0
 529 char *
 530 RepoMeta::BucketList (void)
 531 {
 532         /*
 533          * TBD: make this return values instead of producing output.
 534          * This is just a code fragment showing how to get a list of buckets,
 535          * in case I forget.
 536          */
 537         BSONObj                         repl;
 538
 539         BSONObj dist = BSON("distinct"<<"main"<<"_key"<<"_bucket");
 540         if (client.runCommand("repo",dist,repl)) {
 541                 cout << repl.toString() << endl;
 542                 BSONObj elem = repl.getField("values").embeddedObject();
 543                 for (int i = 0; i < elem.nFields(); ++i) {
 544                         cout << elem[i].str() << endl;
 545                 }
 546         }
 547 }
 548 #endif
 549
 550 void
 551 RepoMeta::Delete (const char *bucket, const char *key)
 552 {
 553         Query   q       = QUERY("_bucket"<<bucket<<"_key"<<key);
 554
 555         try {
 556                 client.remove(MAIN_TBL,q);
 557         }
 558         catch (ConnectException &ce) {
 559                 cerr << "Delete failed for " << bucket << "/" << key << endl;
 560         }
 561 }
 562
 563 extern "C"
 564 void
 565 meta_delete (const char *bucket, const char *key)
 566 {
 567         CLIENT_LOCK;
 568         it->Delete(bucket,key);
 569         CLIENT_UNLOCK;
 570 }
 571
 572 size_t
 573 RepoMeta::GetSize (const char *bucket, const char *key)
 574 {
 575         auto_ptr<DBClientCursor>        curs;
 576         Query                           q;
 577         BSONObj                         bo;
 578         const char *                    data;
 579
 580         (void)data;
 581
 582         q = QUERY("_bucket"<<bucket<<"_key"<<key);
 583         curs = GetCursor(q);
 584
 585         if (!curs->more()) {
 586                 return 0;
 587         }
 588
 589         bo = curs->next();
 590         return bo.getField("_size").numberLong();
 591 }
 592
 593 extern "C"
 594 size_t
 595 meta_get_size (const char *bucket, const char *key)
 596 {
 597         size_t  rc;
 598
 599         CLIENT_LOCK;
 600         rc = it->GetSize(bucket,key);
 601         CLIENT_UNLOCK;
 602
 603         return rc;
 604 }
 605
 606 class AttrList {
 607 public:
 608                                 AttrList        (BSONObj &);
 609         int                     Next            (const char **, const char **);
 610         BSONObj                 obj;
 611         vector<BSONElement>     vec;
 612         int                     idx;
 613 };
 614
 615 AttrList::AttrList (BSONObj &bo)
 616 {
 617         obj = bo.copy();
 618         obj.elems(vec);
 619         idx = 0;
 620 }
 621
 622 int
 623 AttrList::Next (const char **name, const char **value)
 624 {
 625         BSONElement     elem;
 626
 627         while (idx < vec.size()) {
 628                 elem = vec[idx++];
 629                 if (elem.type() == String) {
 630                         *name = elem.fieldName();
 631                         *value = elem.String().c_str();
 632                         return 1;
 633                 }
 634         }
 635
 636         return 0;
 637 }
 638
 639 void *
 640 RepoMeta::GetAttrList (const char *bucket, const char *key)
 641 {
 642         auto_ptr<DBClientCursor>        curs;
 643         Query                           q;
 644         BSONObj                         bo;
 645
 646         q = QUERY("_bucket"<<bucket<<"_key"<<key);
 647         curs = GetCursor(q);
 648
 649         if (!curs->more()) {
 650                 return NULL;
 651         }
 652         bo = curs->next();
 653
 654         return new AttrList(bo);
 655 }
 656 extern "C"
 657 void *
 658 meta_get_attrs (const char *bucket, const char *key)
 659 {
 660         void *poc;
 661
 662         CLIENT_LOCK;
 663         poc = it->GetAttrList(bucket,key);
 664         CLIENT_UNLOCK;
 665
 666         return poc;
 667 }
 668
 669 extern "C"
 670 int
 671 meta_attr_next (void *ctx, const char **name, const char **value)
 672 {
 673         AttrList *poc = (AttrList *)ctx;
 674
 675         return poc->Next(name,value);
 676 }
 677
 678 extern "C"
 679 void
 680 meta_attr_stop (void *ctx)
 681 {
 682         AttrList *poc = (AttrList *)ctx;
 683
 684         delete poc;
 685 }