examples/c/ex_rep_gsg/rep_mgr_gsg.c

   1 /*-
   2  * See the file LICENSE for redistribution information.
   3  *
   4  * Copyright (c) 2006, 2012 Oracle and/or its affiliates.  All rights reserved.
   5  *
   6  * $Id$
   7  */
   8
   9 /*
  10  * NOTE: This example is a simplified version of the rep_mgr.c
  11  * example that can be found in the db/examples/c/ex_rep/mgr directory.
  12  *
  13  * This example is intended only as an aid in learning Replication Manager
  14  * concepts.  It is not complete in that many features are not exercised
  15  * in it, nor are many error conditions properly handled.
  16  *
  17  */
  18
  19 #include <stdlib.h>
  20 #include <string.h>
  21 #include <errno.h>
  22 #ifdef _WIN32
  23 #include <windows.h>
  24 #define sleep(s)                Sleep(1000 * (s))
  25 #else /* !_WIN32 */
  26 #include <unistd.h>
  27 #endif
  28
  29 #include <db.h>
  30
  31 #ifdef _WIN32
  32 extern int getopt(int, char * const *, const char *);
  33 #endif
  34
  35 #define CACHESIZE (10 * 1024 * 1024)
  36 #define DATABASE "quote.db"
  37 #define SLEEPTIME 3
  38
  39 typedef struct {
  40     int is_master;
  41 } APP_DATA;
  42
  43 const char *progname = "ex_rep_gsg_repmgr";
  44
  45 int create_env(const char *, DB_ENV **);
  46 int env_init(DB_ENV *, const char *);
  47 int doloop (DB_ENV *);
  48 int print_stocks(DB *);
  49 static void event_callback(DB_ENV *, u_int32_t, void *);
  50
  51 /* Usage function */
  52 static void
  53 usage()
  54 {
  55     fprintf(stderr, "usage: %s ", progname);
  56     fprintf(stderr, "-h home -l|-L host:port\n");
  57     fprintf(stderr, "\t\t[-r host:port][-p priority]\n");
  58     fprintf(stderr, "where:\n");
  59     fprintf(stderr, "\t-h identifies the environment home directory ");
  60     fprintf(stderr, "(required).\n");
  61     fprintf(stderr, "\t-l identifies the host and port used by this ");
  62     fprintf(stderr, "site (required unless L is specified).\n");
  63     fprintf(stderr, "\t-L identifies the local site as group creator. \n");
  64     fprintf(stderr, "\t-r identifies another site participating in ");
  65     fprintf(stderr, "this replication group\n");
  66     fprintf(stderr, "\t-p identifies the election priority used by ");
  67     fprintf(stderr, "this replica.\n");
  68     exit(EXIT_FAILURE);
  69 }
  70
  71 int
  72 main(int argc, char *argv[])
  73 {
  74     DB_ENV *dbenv;
  75     DB_SITE *dbsite;
  76     extern char *optarg;
  77     const char *home;
  78     char ch, *host, *portstr;
  79     int local_is_set, ret, is_group_creator;
  80     u_int16_t port;
  81     /* Used to track whether this is a replica or a master. */
  82     APP_DATA my_app_data;
  83
  84     dbenv = NULL;
  85     ret = local_is_set = is_group_creator = 0;
  86     home = NULL;
  87
  88     my_app_data.is_master = 0;  /* Assume that we start as a replica */
  89
  90     if ((ret = create_env(progname, &dbenv)) != 0)
  91         goto err;
  92
  93     /* Make APP_DATA available through the environment handle. */
  94     dbenv->app_private = &my_app_data;
  95
  96     /* Default priority is 100. */
  97     dbenv->rep_set_priority(dbenv, 100);
  98     /* Permanent messages require at least one ack. */
  99     dbenv->repmgr_set_ack_policy(dbenv, DB_REPMGR_ACKS_ONE);
 100     /* Give 500 microseconds to receive the ack. */
 101     dbenv->rep_set_timeout(dbenv, DB_REP_ACK_TIMEOUT, 500);
 102
 103     /* Collect the command line options. */
 104     while ((ch = getopt(argc, argv, "h:l:L:p:r:")) != EOF)
 105         switch (ch) {
 106         case 'h':
 107             home = optarg;
 108             break;
 109         /* Set the host and port used by this environment. */
 110         case 'L':
 111             is_group_creator = 1; /* FALLTHROUGH */
 112         case 'l':
 113             host = strtok(optarg, ":");
 114             if ((portstr = strtok(NULL, ":")) == NULL) {
 115                 fprintf(stderr, "Bad host specification.\n");
 116                 goto err;
 117             }
 118             port = (unsigned short)atoi(portstr);
 119             if ((ret =
 120               dbenv->repmgr_site(dbenv, host, port, &dbsite, 0)) != 0){
 121                 fprintf(stderr, "Could not set local address %s:%d.\n",
 122                   host, port);
 123                 goto err;
 124             }
 125             dbsite->set_config(dbsite, DB_LOCAL_SITE, 1);
 126             if (is_group_creator)
 127                 dbsite->set_config(dbsite, DB_GROUP_CREATOR, 1);
 128
 129             if ((ret = dbsite->close(dbsite)) != 0) {
 130                 dbenv->err(dbenv, ret, "DB_SITE->close");
 131                 goto err;
 132         }
 133             local_is_set = 1;
 134             break;
 135         /* Set this replica's election priority. */
 136         case 'p':
 137             dbenv->rep_set_priority(dbenv, atoi(optarg));
 138             break;
 139         /* Identify another site in the replication group. */
 140         case 'r':
 141             host = strtok(optarg, ":");
 142             if ((portstr = strtok(NULL, ":")) == NULL) {
 143                 fprintf(stderr, "Bad host specification.\n");
 144                 goto err;
 145             }
 146             port = (unsigned short)atoi(portstr);
 147             if ((ret = dbenv->repmgr_site(dbenv, host, port, &dbsite, 0)) != 0) {
 148                 dbenv->err(dbenv, ret, "DB_ENV->repmgr_site");
 149                 goto err;
 150             }
 151             dbsite->set_config(dbsite, DB_BOOTSTRAP_HELPER, 1);
 152             if ((ret = dbsite->close(dbsite)) != 0) {
 153                 dbenv->err(dbenv, ret, "DB_SITE->close");
 154                 goto err;
 155             }
 156             break;
 157         case '?':
 158         default:
 159             usage();
 160         }
 161
 162     /* Error check command line. */
 163     if (home == NULL || !local_is_set)
 164         usage();
 165
 166     if ((ret = env_init(dbenv, home)) != 0)
 167         goto err;
 168
 169     if ((ret = dbenv->repmgr_start(dbenv, 3, DB_REP_ELECTION)) != 0)
 170         goto err;
 171
 172     if ((ret = doloop(dbenv)) != 0) {
 173         dbenv->err(dbenv, ret, "Application failed");
 174         goto err;
 175     }
 176
 177 err: if (dbenv != NULL)
 178         (void)dbenv->close(dbenv, 0);
 179
 180     return (ret);
 181
 182 }
 183
 184 /* Create and configure an environment handle. */
 185 int
 186 create_env(const char *progname, DB_ENV **dbenvp)
 187 {
 188     DB_ENV *dbenv;
 189     int ret;
 190
 191     if ((ret = db_env_create(&dbenv, 0)) != 0) {
 192         fprintf(stderr, "can't create env handle: %s\n",
 193             db_strerror(ret));
 194         return (ret);
 195     }
 196
 197     dbenv->set_errfile(dbenv, stderr);
 198     dbenv->set_errpfx(dbenv, progname);
 199     (void)dbenv->set_event_notify(dbenv, event_callback);
 200
 201     *dbenvp = dbenv;
 202     return (0);
 203 }
 204
 205 /* Open and configure an environment. */
 206 int
 207 env_init(DB_ENV *dbenv, const char *home)
 208 {
 209     u_int32_t flags;
 210     int ret;
 211
 212     (void)dbenv->set_cachesize(dbenv, 0, CACHESIZE, 0);
 213     (void)dbenv->set_flags(dbenv, DB_TXN_NOSYNC, 1);
 214
 215     flags = DB_CREATE |
 216         DB_INIT_LOCK |
 217         DB_INIT_LOG |
 218         DB_INIT_MPOOL |
 219         DB_INIT_REP |
 220         DB_INIT_TXN |
 221         DB_RECOVER |
 222         DB_THREAD;
 223     if ((ret = dbenv->open(dbenv, home, flags, 0)) != 0)
 224         dbenv->err(dbenv, ret, "can't open environment");
 225     return (ret);
 226 }
 227
 228 /*
 229  * A callback used to determine whether the local environment is a
 230  * replica or a master. This is called by the Replication Manager
 231  * when the local environment changes state.
 232  */
 233 static void
 234 event_callback(DB_ENV *dbenv, u_int32_t which, void *info)
 235 {
 236     APP_DATA *app = dbenv->app_private;
 237
 238     info = NULL;                /* Currently unused. */
 239
 240     switch (which) {
 241     case DB_EVENT_REP_MASTER:
 242         app->is_master = 1;
 243         break;
 244
 245     case DB_EVENT_REP_CLIENT:
 246         app->is_master = 0;
 247         break;
 248
 249     case DB_EVENT_REP_STARTUPDONE: /* FALLTHROUGH */
 250     case DB_EVENT_REP_NEWMASTER:
 251         /* Ignore. */
 252         break;
 253
 254     default:
 255         dbenv->errx(dbenv, "ignoring event %d", which);
 256     }
 257 }
 258
 259 /*
 260  * Provides the main data processing function for our application.
 261  * This function provides a command line prompt to which the user
 262  * can provide a ticker string and a stock price.  Once a value is
 263  * entered to the application, the application writes the value to
 264  * the database and then displays the entire database.
 265  */
 266 #define    BUFSIZE 1024
 267
 268 int
 269 doloop(DB_ENV *dbenv)
 270 {
 271     DB *dbp;
 272     APP_DATA *app_data;
 273     DBT key, data;
 274     char buf[BUFSIZE], *rbuf;
 275     int ret;
 276     u_int32_t flags;
 277
 278     dbp = NULL;
 279     ret = 0;
 280     memset(&key, 0, sizeof(key));
 281     memset(&data, 0, sizeof(data));
 282     app_data = dbenv->app_private;
 283
 284     for (;;) {
 285         if (dbp == NULL) {
 286             if ((ret = db_create(&dbp, dbenv, 0)) != 0)
 287                 return (ret);
 288
 289             flags = DB_AUTO_COMMIT;
 290             if (app_data->is_master)
 291                 flags |= DB_CREATE;
 292             if ((ret = dbp->open(dbp,
 293                 NULL, DATABASE, NULL, DB_BTREE, flags, 0)) != 0) {
 294                 if (ret == ENOENT) {
 295                     printf(
 296                       "No stock database yet available.\n");
 297                     if ((ret = dbp->close(dbp, 0)) != 0) {
 298                         dbenv->err(dbenv, ret, "DB->close");
 299                         goto err;
 300                     }
 301                     dbp = NULL;
 302                     sleep(SLEEPTIME);
 303                     continue;
 304                 }
 305                 dbenv->err(dbenv, ret, "DB->open");
 306                 goto err;
 307             }
 308         }
 309
 310         printf("QUOTESERVER%s> ",
 311             app_data->is_master ? "" : " (read-only)");
 312         fflush(stdout);
 313
 314         if (fgets(buf, sizeof(buf), stdin) == NULL)
 315             break;
 316         if (strtok(&buf[0], " \t\n") == NULL) {
 317             switch ((ret = print_stocks(dbp))) {
 318             case 0:
 319                 continue;
 320             case DB_REP_HANDLE_DEAD:
 321                 (void)dbp->close(dbp, DB_NOSYNC);
 322                 dbp = NULL;
 323                 dbenv->errx(dbenv, "Got a dead replication handle");
 324                 continue;
 325             default:
 326                 dbp->err(dbp, ret, "Error traversing data");
 327                 goto err;
 328             }
 329         }
 330         rbuf = strtok(NULL, " \t\n");
 331         if (rbuf == NULL || rbuf[0] == '\0') {
 332             if (strncmp(buf, "exit", 4) == 0 ||
 333                 strncmp(buf, "quit", 4) == 0)
 334                 break;
 335             dbenv->errx(dbenv, "Format: TICKER VALUE");
 336             continue;
 337         }
 338
 339         if (!app_data->is_master) {
 340             dbenv->errx(dbenv, "Can't update at client");
 341             continue;
 342         }
 343
 344         key.data = buf;
 345         key.size = (u_int32_t)strlen(buf);
 346
 347         data.data = rbuf;
 348         data.size = (u_int32_t)strlen(rbuf);
 349
 350         if ((ret = dbp->put(dbp,
 351             NULL, &key, &data, 0)) != 0) {
 352             dbp->err(dbp, ret, "DB->put");
 353             goto err;
 354         }
 355     }
 356
 357 err: if (dbp != NULL)
 358         (void)dbp->close(dbp, DB_NOSYNC);
 359
 360     return (ret);
 361 }
 362
 363 /* Display all the stock quote information in the database. */
 364 int
 365 print_stocks(DB *dbp)
 366 {
 367     DBC *dbc;
 368     DBT key, data;
 369 #define    MAXKEYSIZE    10
 370 #define    MAXDATASIZE    20
 371     char keybuf[MAXKEYSIZE + 1], databuf[MAXDATASIZE + 1];
 372     int ret, t_ret;
 373     u_int32_t keysize, datasize;
 374
 375     if ((ret = dbp->cursor(dbp, NULL, &dbc, 0)) != 0) {
 376         dbp->err(dbp, ret, "can't open cursor");
 377         return (ret);
 378     }
 379
 380     memset(&key, 0, sizeof(key));
 381     memset(&data, 0, sizeof(data));
 382
 383     printf("\tSymbol\tPrice\n");
 384     printf("\t======\t=====\n");
 385
 386     for (ret = dbc->c_get(dbc, &key, &data, DB_FIRST);
 387         ret == 0;
 388         ret = dbc->c_get(dbc, &key, &data, DB_NEXT)) {
 389         keysize = key.size > MAXKEYSIZE ? MAXKEYSIZE : key.size;
 390         memcpy(keybuf, key.data, keysize);
 391         keybuf[keysize] = '\0';
 392
 393         datasize = data.size >= MAXDATASIZE ? MAXDATASIZE : data.size;
 394         memcpy(databuf, data.data, datasize);
 395         databuf[datasize] = '\0';
 396
 397         printf("\t%s\t%s\n", keybuf, databuf);
 398     }
 399     printf("\n");
 400     fflush(stdout);
 401
 402     if ((t_ret = dbc->c_close(dbc)) != 0 && ret == 0)
 403         ret = t_ret;
 404
 405     switch (ret) {
 406     case 0:
 407     case DB_NOTFOUND:
 408         return (0);
 409     case DB_LOCK_DEADLOCK:
 410         return (0);
 411     default:
 412         return (ret);
 413     }
 414 }
 415