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         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
  84         char *  DidPut          (const char *bucket, const char *key,
  85                                  const char *loc, size_t size);
  86         void    GotCopy         (const char *bucket, const char *key,
  87                                  const char *loc);
  88         char *  HasCopy         (const char *bucket, const char *key,
  89                                  const char *loc);
  90         int     SetValue        (const char *bucket, const char *key,
  91                                  const char *mkey, const char * mvalue);
  92         int     GetValue        (const char *bucket, const char *key,
  93                                  const char *mkey, char ** mvalue);
  94         RepoQuery * NewQuery    (const char *bucket, const char *key,
  95                                  const char * expr);
  96         auto_ptr<DBClientCursor> GetCursor (Query &q);
  97         void    Delete          (const char *bucket, const char *key);
  98         size_t  GetSize         (const char *bucket, const char *key);
  99         int     Check           (const char *bucket, const char *key,
 100                                  const char *depot);
 101 };
 102
 103 class RepoQuery {
 104         RepoMeta &              parent;
 105         DBClientCursor *        curs;
 106         value_t *               expr;
 107 public:
 108                 RepoQuery       (const char *, const char *, const char *,
 109                                  RepoMeta &);
 110                 ~RepoQuery      ();
 111         bool    Next            (void);
 112         char    *bucket;
 113         char    *key;
 114         getter_t getter;
 115 };
 116
 117 RepoMeta *it;
 118
 119 RepoMeta::RepoMeta ()
 120 {
 121         char    addr[128];
 122
 123         sprintf(addr,"%s:%u",db_host,db_port);
 124         client.connect(addr);
 125 }
 126
 127 extern "C" void
 128 meta_init (void)
 129 {
 130         it = new RepoMeta();
 131 }
 132
 133 RepoMeta::~RepoMeta ()
 134 {
 135 }
 136
 137 extern "C" void
 138 meta_fini (void)
 139 {
 140         delete it;
 141 }
 142
 143 auto_ptr<DBClientCursor>
 144 RepoMeta::GetCursor (Query &q)
 145 {
 146         auto_ptr<DBClientCursor> curs;
 147
 148         curs = client.query(MAIN_TBL,q);
 149         if (!curs.get()) {
 150                 cout << "reconnecting" << endl;
 151                 try {
 152                         client.connect("localhost");
 153                 }
 154                 catch (ConnectException &ce) {
 155                         cout << "server down" << endl;
 156                         throw;
 157                 }
 158                 curs = client.query(MAIN_TBL,q);
 159         }
 160
 161         return curs;
 162 }
 163
 164 char *
 165 RepoMeta::DidPut (const char *bucket, const char *key, const char *loc,
 166                   size_t size)
 167 {
 168         BSONObjBuilder                  bb;
 169         struct timeval                  now_tv;
 170         double                          now;
 171         auto_ptr<DBClientCursor>        curs;
 172         Query                           q;
 173         char                            now_str[sizeof(now)*2+1];
 174
 175         gettimeofday(&now_tv,NULL);
 176         now = (double)now_tv.tv_sec + (double)now_tv.tv_usec / 1000000.0;
 177         dbl_to_str(&now,now_str);
 178         cout << "now_str = " << now_str << endl;
 179
 180         q = QUERY("bucket"<<bucket<<"key"<<key);
 181         curs = GetCursor(q);
 182         if (curs->more()) {
 183                 /* Nice functionality, but what an ugly syntax! */
 184                 client.update(MAIN_TBL,q,BSON(
 185                         "$set"<<BSON("loc"<<BSON_ARRAY(loc))
 186                 <<      "$set"<<BSON("date"<<now)
 187                 <<      "$set"<<BSON("etag"<<now_str)
 188                 <<      "$set"<<BSON("size"<<(long long)size)));
 189 #if 0
 190                 client.update(MAIN_TBL,q,
 191                         BSON("$set"<<BSON("loc"<<BSON_ARRAY(loc))));
 192                 client.update(MAIN_TBL,q,
 193                         BSON("$set"<<BSON("date"<<now)));
 194                 client.update(MAIN_TBL,q,
 195                         BSON("$set"<<BSON("etag"<<now_str)));
 196                 client.update(MAIN_TBL,q,
 197                         BSON("$set"<<BSON("size"<<(long long)size)));
 198 #endif
 199         }
 200         else {
 201                 bb << "bucket" << bucket << "key" << key
 202                    << "loc" << BSON_ARRAY(loc) << "date" << now
 203                    << "etag" << now_str << "size" << (long long)size;
 204                 client.insert(MAIN_TBL,bb.obj());
 205         }
 206
 207         return strdup(now_str);
 208 }
 209
 210 extern "C" char *
 211 meta_did_put (const char *bucket, const char *key, const char *loc, size_t size)
 212 {
 213         char    *rc;
 214
 215         cout << "meta_did_put(" << bucket << "," << key << "," << loc << ")"
 216              << endl;
 217
 218         CLIENT_LOCK;
 219         rc = it->DidPut(bucket,key,loc,size);
 220         CLIENT_UNLOCK;
 221
 222         return rc;
 223 }
 224
 225 void
 226 RepoMeta::GotCopy (const char *bucket, const char *key, const char *loc)
 227 {
 228         BSONObjBuilder                  bb;
 229         auto_ptr<DBClientCursor>        curs;
 230         Query                           q;
 231
 232         q = QUERY("bucket"<<bucket<<"key"<<key);
 233         curs = GetCursor(q);
 234         if (curs->more()) {
 235                 /* Nice functionality, but what an ugly syntax! */
 236                 client.update(MAIN_TBL,q,BSON("$addToSet"<<BSON("loc"<<loc)));
 237         }
 238         else {
 239                 cerr << bucket << ":" << key << " not found in GotCopy!" << endl;
 240         }
 241 }
 242
 243 extern "C" void
 244 meta_got_copy (const char *bucket, const char *key, const char *loc)
 245 {
 246         CLIENT_LOCK;
 247         it->GotCopy(bucket,key,loc);
 248         CLIENT_UNLOCK;
 249 }
 250
 251 char *
 252 RepoMeta::HasCopy (const char *bucket, const char *key, const char *loc)
 253 {
 254         BSONObjBuilder                  bb;
 255         auto_ptr<DBClientCursor>        curs;
 256         Query                           q;
 257         const char                      *value;
 258
 259         q = QUERY("bucket"<<bucket<<"key"<<key<<"loc"<<loc);
 260         curs = GetCursor(q);
 261         if (!curs->more()) {
 262                 cout << bucket << "/" << key << " not found at " << loc << endl;
 263                 return NULL;
 264         }
 265
 266         value = curs->next().getStringField("etag");
 267         if (!value || !*value) {
 268                 cout << bucket << "/" << key << " no etag at " << loc << endl;
 269                 return NULL;
 270         }
 271
 272         cout << bucket << "/" << key << " etag = " << value << endl;
 273         return strdup(value);
 274 }
 275
 276 extern "C" char *
 277 meta_has_copy (const char *bucket, const char *key, const char *loc)
 278 {
 279         char    *rc;
 280
 281         CLIENT_LOCK;
 282         rc = it->HasCopy(bucket,key,loc);
 283         CLIENT_UNLOCK;
 284
 285         return rc;
 286 }
 287
 288 int
 289 RepoMeta::SetValue (const char *bucket, const char *key, const char *mkey,
 290                     const char * mvalue)
 291 {
 292         Query   q       = QUERY("bucket"<<bucket<<"key"<<key);
 293
 294         client.update(MAIN_TBL,q,BSON("$set"<<BSON(mkey<<mvalue)),1);
 295         // TBD: check for and propagate errors.
 296         return 0;
 297 }
 298
 299 extern "C" int
 300 meta_set_value (const char *bucket, const char *key, const char *mkey,
 301                 const char * mvalue)
 302 {
 303         int     rc;
 304
 305         CLIENT_LOCK;
 306         rc = it->SetValue(bucket,key,mkey,mvalue);
 307         CLIENT_UNLOCK;
 308
 309         return rc;
 310 }
 311
 312 int
 313 RepoMeta::GetValue (const char *bucket, const char *key, const char *mkey,
 314                     char ** mvalue)
 315 {
 316         auto_ptr<DBClientCursor>        curs;
 317         Query                           q;
 318         BSONObj                         bo;
 319         const char *                    data;
 320
 321         q = QUERY("bucket"<<bucket<<"key"<<key);
 322         curs = GetCursor(q);
 323
 324         if (!curs->more()) {
 325                 return ENXIO;
 326         }
 327
 328         bo = curs->next();
 329         data = bo.getStringField(mkey);
 330         if (!data || !*data) {
 331                 return ENXIO;
 332         }
 333
 334         *mvalue = strdup(data);
 335         return 0;
 336 }
 337
 338 extern "C" int
 339 meta_get_value (const char *bucket, const char *key, const char *mkey,
 340                 char ** mvalue)
 341 {
 342         int     rc;
 343
 344         CLIENT_LOCK;
 345         rc = it->GetValue(bucket,key,mkey,mvalue);
 346         CLIENT_UNLOCK;
 347
 348         return rc;
 349 }
 350
 351 RepoQuery::RepoQuery (const char *bucket, const char *key, const char *qstr,
 352                       RepoMeta &p)
 353         : parent(p)
 354 {
 355         Query                           q;
 356         auto_ptr<DBClientCursor>        tmp;
 357
 358         if (bucket) {
 359                 cout << "bucket is " << bucket << " and we don't care" << endl;
 360                 q = QUERY("bucket"<<bucket);
 361         }
 362         else if (key) {
 363                 cout << "key is " << key << " and we don't care" << endl;
 364                 q = QUERY("key"<<key);
 365         }
 366         else {
 367                 abort();
 368         }
 369
 370         /*
 371          * TBD: we should really convert our query into one of Mongo's,
 372          * and let them do all the work.  Handling the general case
 373          * would be pretty messy, but we could handle specific cases
 374          * pretty easily.  For example, a very high percentage of
 375          * queries are likely to be a single field/value comparison.
 376          * For now just punt, but revisit later.
 377          */
 378
 379         if (qstr) {
 380                 expr = parse(qstr);
 381                 if (expr) {
 382                         print_value(expr);
 383                 }
 384                 else {
 385                         cout << "could not parse " << qstr << endl;
 386                 }
 387         }
 388         else {
 389                 expr = NULL;
 390         }
 391
 392         curs = parent.GetCursor(q).release();
 393         bucket = NULL;
 394         key = NULL;
 395 }
 396
 397 RepoQuery::~RepoQuery ()
 398 {
 399         cout << "in " << __func__ << endl;
 400         if (expr) {
 401                 free_value(expr);
 402         }
 403
 404         delete curs;
 405 }
 406
 407 extern "C" void
 408 meta_query_stop (void * qobj)
 409 {
 410         CLIENT_LOCK;
 411         delete (RepoQuery *)qobj;
 412         CLIENT_UNLOCK;
 413 }
 414
 415 extern "C" char *
 416 query_getter (void *ctx, const char *id)
 417 {
 418         BSONObj *cur_bo = (BSONObj *)ctx;
 419
 420         return (char *)cur_bo->getStringField(id);
 421 }
 422
 423 bool
 424 RepoQuery::Next (void)
 425 {
 426         BSONObj         bo;
 427
 428         while (curs->more()) {
 429                 bo = curs->next();
 430                 if (expr) {
 431                         getter.func = query_getter;
 432                         getter.ctx = (void *)&bo;
 433                         if (eval(expr,&getter,NULL) <= 0) {
 434                                 continue;
 435                         }
 436                 }
 437                 bucket = (char *)bo.getStringField("bucket");
 438                 key = (char *)bo.getStringField("key");
 439                 return true;
 440         }
 441
 442         return false;
 443 }
 444
 445 RepoQuery *
 446 RepoMeta::NewQuery (const char *bucket, const char *key, const char *expr)
 447 {
 448         return new RepoQuery(bucket,key,expr,*this);
 449 }
 450
 451 extern "C" void *
 452 meta_query_new (const char *bucket, const char *key, const char *expr)
 453 {
 454         void    *rc;
 455
 456         if ((bucket && key) || (!bucket && !key)) {
 457                 return NULL;
 458         }
 459
 460         CLIENT_LOCK;
 461         rc = it->NewQuery(bucket,key,expr);
 462         CLIENT_UNLOCK;
 463
 464         return rc;
 465 }
 466
 467 extern "C" int
 468 meta_query_next (void * qobj, char ** bucket, char ** key)
 469 {
 470         RepoQuery *     rq      = (RepoQuery *)qobj;
 471
 472         CLIENT_LOCK;
 473         if (!rq->Next()) {
 474                 CLIENT_UNLOCK;
 475                 return 0;
 476         }
 477         CLIENT_UNLOCK;
 478
 479         *bucket = rq->bucket;
 480         *key = rq->key;
 481         return 1;
 482 }
 483
 484 #if 0
 485 char *
 486 RepoMeta::BucketList (void)
 487 {
 488         /*
 489          * TBD: make this return values instead of producing output.
 490          * This is just a code fragment showing how to get a list of buckets,
 491          * in case I forget.
 492          */
 493         BSONObj                         repl;
 494
 495         BSONObj dist = BSON("distinct"<<"main"<<"key"<<"bucket");
 496         if (client.runCommand("repo",dist,repl)) {
 497                 cout << repl.toString() << endl;
 498                 BSONObj elem = repl.getField("values").embeddedObject();
 499                 for (int i = 0; i < elem.nFields(); ++i) {
 500                         cout << elem[i].str() << endl;
 501                 }
 502         }
 503 }
 504 #endif
 505
 506 void
 507 RepoMeta::Delete (const char *bucket, const char *key)
 508 {
 509         Query   q       = QUERY("bucket"<<bucket<<"key"<<key);
 510
 511         client.remove(MAIN_TBL,q);
 512 }
 513
 514 extern "C"
 515 void
 516 meta_delete (const char *bucket, const char *key)
 517 {
 518         CLIENT_LOCK;
 519         it->Delete(bucket,key);
 520         CLIENT_UNLOCK;
 521 }
 522
 523 size_t
 524 RepoMeta::GetSize (const char *bucket, const char *key)
 525 {
 526         auto_ptr<DBClientCursor>        curs;
 527         Query                           q;
 528         BSONObj                         bo;
 529         const char *                    data;
 530
 531         q = QUERY("bucket"<<bucket<<"key"<<key);
 532         curs = GetCursor(q);
 533
 534         if (!curs->more()) {
 535                 return 0;
 536         }
 537
 538         bo = curs->next();
 539         return bo.getField("size").numberLong();
 540 }
 541
 542 extern "C"
 543 size_t
 544 meta_get_size (const char *bucket, const char *key)
 545 {
 546         size_t  rc;
 547
 548         CLIENT_LOCK;
 549         rc = it->GetSize(bucket,key);
 550         CLIENT_UNLOCK;
 551
 552         return rc;
 553 }