JSOC_Documentation: base/drms/libs/main/jsoc

00001 
00170 #include "drms.h"
00171 #include "serverdefs.h"
00172 #include "xmem.h"
00173 #include "cmdparams.h"
00174 #include "cmdparams_priv.h"
00175 #include "tee.h"
00176 #include "jsoc_main.h"
00177 #ifdef __linux__
00178 #include "backtrace.h"
00179 #include <sched.h>
00180 #endif
00181 
00185 DRMS_Env_t *drms_env;
00186 
00191 CmdParams_t cmdparams;
00195 ModuleArgs_t *gModArgs = module_args;
00196 
00203 CmdParams_t *GetGlobalCmdParams() 
00204 {
00205   return &cmdparams;
00206 }
00207 
00208 static void atexit_action (void) {
00209   fprintf (stderr, "WARNING: DRMS server called exit.\n");
00210 #ifdef DEBUG
00211 #ifdef __linux__
00212   fprintf (stderr, "Stack frame:\n");
00213   show_stackframe (stderr);
00214   hang (12);
00215 #endif
00216 #endif
00217   drms_server_abort (drms_env, 1);
00218 #ifdef DEBUG
00219    fprintf (stderr, "drms_server_abort() call completed.\n");
00220    xmem_leakreport ();
00221 #endif
00222 }
00223 
00224 /* db_lock calls this function to determine what signals it should NOT 
00225  * allow to interrupt code running inside the db lock. */
00226 void drms_createsigmask(sigset_t *set, void *data)
00227 {
00228    pthread_t threadid = *((pthread_t *)data);
00229    if (set && threadid == pthread_self())
00230    {
00231       sigemptyset(set);
00232       sigaddset(set, SIGUSR2); 
00233    }
00234 }
00235 
00236 /* If shutdown has started, waits until DoIt() notified */
00237 static DRMS_Shutdown_State_t GetShutdownState(DRMS_Env_t *env)
00238 {
00239    DRMS_Shutdown_State_t st = kSHUTDOWN_UNINITIATED;
00240    sem_t *sdsem = drms_server_getsdsem();
00241 
00242    if (sdsem)
00243    {
00244       /* Wait until the signal thread is not messing with shutdown - if it is, then 
00245        * main will block here until the signal thread terminates the process */
00246       sem_wait(sdsem);
00247 
00248       DRMS_Shutdown_State_t sdstate = drms_server_getsd();
00249 
00250       if (sdstate == kSHUTDOWN_UNINITIATED)
00251       {
00252          sem_post(sdsem);
00253       }
00254       else if (sdstate == kSHUTDOWN_COMMIT || sdstate == kSHUTDOWN_ABORT || sdstate == kSHUTDOWN_BYMAIN)
00255       {
00256          /* signal-induced shutdown has already been started, and DoIt() already knows about it, 
00257           * or main thread is shutting down. */
00258          st = sdstate;
00259          sem_post(sdsem);
00260       }
00261       else
00262       {
00263          /* shutdown initiated by signal thread, but DoIt() not notified yet */
00264          sem_post(sdsem);
00265 
00266          /* call commit or abort as appropriate.
00267           * sdstate might indicate DoIt() not notified yet. */
00268          while (1)
00269          {
00270             sem_wait(sdsem);
00271             sdstate = drms_server_getsd();
00272 
00273             if (sdstate == kSHUTDOWN_ABORT || sdstate == kSHUTDOWN_COMMIT)
00274             {
00275                st = sdstate;
00276                sem_post(sdsem);
00277                break;
00278             }
00279 
00280             sem_post(sdsem);
00281             sleep(1);
00282          }
00283       }
00284    }
00285 
00286    return st;
00287 }
00288 
00289 static void FinalBits(DRMS_Env_t *env, int abort_flag)
00290 {
00291    /* Acquire shutdown lock, to prevent the signal thread from trying to simultaneously shut down. */
00292    /* If signal thread has the lock, then main waits here until the process dies. */
00293    sem_t *sdsem = drms_server_getsdsem();
00294 
00295    if (sdsem)
00296    {
00297       sem_wait(sdsem);
00298    }
00299 
00300    if (drms_server_getsd() != kSHUTDOWN_UNINITIATED)
00301    {
00302       /* signal thread is already shutting down, just wait for signal thread to finish 
00303        * (which it won't do, because it is going to call exit, so this is an indefinite sleep). */
00304       sem_post(sdsem);
00305       pthread_join(env->signal_thread, NULL);  
00306    }
00307    else
00308    {
00309       /* Tell signal thread not to accept shutdown requests, because main is shutting down. */
00310       drms_server_setsd(kSHUTDOWN_BYMAIN);
00311 
00312       sem_post(sdsem);
00313 
00314       /* Now, signal thread will not try and shut down. */
00315 
00316       /* Calls drms_server_commit() or drms_server_abort().
00317        * Don't set last argment to final - still need environment below. */
00318       /* This will cause the SUMS thread to return. */
00319       /* If no transaction actually started, then noop. */
00320       drms_server_end_transaction(env, abort_flag, 0);
00321 
00322 #ifndef DEBUG
00323       /* Clean up signal thread - the signal hasn't received a shutdown signal, so it is still
00324        * running. Send USR2 to tell signal thread to end itself. */
00325       pthread_kill(env->signal_thread, SIGUSR2);
00326       pthread_join(env->signal_thread, NULL);
00327 #endif
00328 
00329       /* Must disconnect db, because we didn't set the final flag in drms_server_end_transaction() */
00330       db_disconnect(&env->session->db_handle);
00331 
00332       /* Free cmd-params (valgrind reports this - let's just clean up so it doesn't show up on 
00333        * valgrind's radar). */
00334       cmdparams_freeall(&cmdparams);
00335 
00336       drms_free_env(env, 1); 
00337 
00338       /* doesn't call at_exit_action() */
00339       _exit(abort_flag);
00340    }
00341 }
00342 
00343 static int FreeCmdparams(void *data)
00344 {
00345    cmdparams_freeall(&cmdparams);
00346    memset(&cmdparams, 0, sizeof(CmdParams_t));
00347    return 0;
00348 }
00349 
00357 int JSOCMAIN_Main(int argc, char **argv, const char *module_name, int (*CallDoIt)(void)) {
00358   int verbose=0, dolog=0, quiet=0;
00359   int status;
00360   char *user, unknown[]="unknown";
00361   DB_Handle_t *db_handle;
00362     const char *dbhost;
00363     char *dbuser, *dbpasswd, *dbname, *sessionns;
00364     char *dbport = NULL;
00365     char dbHostAndPort[64];
00366   int printrel = 0;
00367   char reservebuf[256];
00368   int16_t retention;
00369   int16_t newsuretention;
00370 
00371    /* Initialize globals here */
00372    memset(&cmdparams, 0, sizeof(CmdParams_t));
00373 
00374 #ifdef DEBUG
00375   xmem_config(1,1,1,1,1000000,1,0,0); 
00376 #endif
00377   /* Parse command line parameters. */
00378   snprintf(reservebuf, 
00379            sizeof(reservebuf), 
00380            "%s,%s,%s,%s,%s,%s,%s,%s",
00381            "L,Q,V,jsocmodver", 
00382            kARCHIVEARG,
00383            kRETENTIONARG,
00384            kNewSuRetention,
00385            kQUERYMEMARG,
00386            kLoopConn,
00387            kDBTimeOut,
00388            kCreateShadows,
00389            kDBUtf8ClientEncoding);
00390   cmdparams_reserve(&cmdparams, reservebuf, "jsocmain");
00391 
00392   status = cmdparams_parse (&cmdparams, argc, argv);
00393   if (status == CMDPARAMS_QUERYMODE) {
00394     cmdparams_usage (argv[0]);
00395     return 0;
00396   } else if (status == CMDPARAMS_NODEFAULT) {
00397     fprintf (stderr, "For usage, type %s [-H|--help]\n", argv[0]);
00398     return 0;
00399   } else if (status < 0) {
00400     fprintf (stderr, "Error: Command line parsing failed. Aborting.\n");
00401     fprintf (stderr, "For usage, type %s [-H|--help]\n", argv[0]);
00402     return 1;
00403   }
00404 
00405   printrel = cmdparams_isflagset(&cmdparams, "jsocmodver");
00406 
00407   if (printrel)
00408   {
00409      char verstr[32];
00410      int isdev = 0;
00411 
00412      jsoc_getversion(verstr, sizeof(verstr), &isdev);
00413      fprintf(stdout, 
00414              "Module '%s' JSOC version is '%s' (%s)\n", 
00415              module_name, 
00416              verstr, 
00417              isdev ? "development" : "release");
00418      return 0;
00419   }
00420 
00421   verbose = (cmdparams_exists (&cmdparams, "V") &&
00422          cmdparams_get_int (&cmdparams, "V", NULL) != 0);
00423   if (verbose) cmdparams_printall (&cmdparams);
00424   quiet = (cmdparams_exists (&cmdparams, "Q") &&
00425        cmdparams_get_int (&cmdparams, "Q", NULL) != 0);
00426   dolog = (cmdparams_exists (&cmdparams, "L") &&
00427        cmdparams_get_int (&cmdparams, "L", NULL) != 0);
00428 
00429   /* Get user name */
00430   if (!(user = getenv("USER")))
00431     user = unknown;
00432 
00433   /* Get hostname, user, passwd and database name for establishing 
00434      a connection to the DRMS database server. */
00435     
00436     /* SERVER does not contain port information. Yet when dbhost is used in db_connect(), that function
00437      * parses the value looking for an optional port number. So if you didn't provide the JSOC_DBHOST
00438      * then there was no way to connect to the db with a port other than the default port that the
00439      * db listens on for incoming connections (which is usually 5432).
00440      *
00441      * I changed this so that masterlists uses the DRMSPGPORT macro to define the port to connect to.
00442      * If by chance somebody has appeneded the port number to the server name in SERVER, and that
00443      * conflicts with the value in DRMSPGPORT, then DRMSPGPORT wins, and a warning message is printed.
00444      *
00445      * --ART (2014.08.20)
00446      */
00447     
00448   if ((dbhost = cmdparams_get_str(&cmdparams, "JSOC_DBHOST", NULL)) == NULL)
00449     {
00450         const char *sep = NULL;
00451         
00452         dbhost =  SERVER;
00453         dbport = DRMSPGPORT;
00454         
00455         /* Check for conflicting port numbers. */
00456         if ((sep = strchr(dbhost, ':')) != NULL)
00457         {
00458             if (strcmp(sep + 1, dbport) != 0)
00459             {
00460                 char *tmpBuf = strdup(dbhost);
00461                 
00462                 if (tmpBuf)
00463                 {
00464                     tmpBuf[sep - dbhost] = '\0';
00465                     fprintf(stderr, "WARNING: the port number in the SERVER localization parameter (%s) and in DRMSPGPORT (%s) conflict.\nThe DRMSPGPORT value will be used.\n", sep + 1, DRMSPGPORT);
00466                     
00467                     snprintf(dbHostAndPort, sizeof(dbHostAndPort), "%s:%s", tmpBuf, dbport);
00468                     free(tmpBuf);
00469                     tmpBuf = NULL;
00470                 }
00471                 else
00472                 {
00473                     fprintf(stderr, "Out of memory.\n");
00474                     return 1;
00475                 }
00476             }
00477             else
00478             {
00479                 snprintf(dbHostAndPort, sizeof(dbHostAndPort), "%s", dbhost);
00480             }
00481         }
00482         else
00483         {
00484             snprintf(dbHostAndPort, sizeof(dbHostAndPort), "%s:%s", dbhost, dbport);
00485         }
00486     }
00487     else
00488     {
00489         snprintf(dbHostAndPort, sizeof(dbHostAndPort), "%s", dbhost);
00490     }
00491 
00492   if ((dbname = cmdparams_get_str(&cmdparams, "JSOC_DBNAME", NULL)) == NULL)
00493     dbname = DBNAME;
00494   if ((dbuser = cmdparams_get_str(&cmdparams, "JSOC_DBUSER", NULL)) == NULL)
00495     dbuser = USER;
00496   if ((dbpasswd = cmdparams_get_str(&cmdparams, "JSOC_DBPASSWD", NULL)) == NULL)
00497     dbpasswd = PASSWD;
00498   sessionns = cmdparams_get_str(&cmdparams, "JSOC_SESSIONNS", NULL);
00499   if (sessionns && !strcmp(sessionns, "public")) {
00500     fprintf(stderr, "Can't run drms_server in public namespace\n");
00501     return 1;
00502   }
00503 
00504   int archive = INT_MIN;
00505   if (drms_cmdparams_exists(&cmdparams, kARCHIVEARG)) {
00506      archive = drms_cmdparams_get_int(&cmdparams, kARCHIVEARG, NULL);
00507      if (archive != -1 && archive != 0 && archive != 1)
00508      {
00509         archive = INT_MIN;
00510      }
00511   }
00512     
00513     retention = INT16_MIN;
00514     char errbuf[128];
00515     
00516     if (drms_cmdparams_exists(&cmdparams, kRETENTIONARG))
00517     {
00518         retention = drms_cmdparams_get_int16(&cmdparams, kRETENTIONARG, &status);
00519         if (status != DRMS_SUCCESS)
00520         {
00521             if (status == DRMS_ERROR_INVALIDCMDARGCONV)
00522             {
00523                 snprintf(errbuf, sizeof(errbuf), "The value for %s must be a 15-bit positive integer.", kRETENTIONARG);
00524                 fprintf(stderr, errbuf);
00525             }
00526             
00527             snprintf(errbuf, sizeof(errbuf), "Invalid value for %s.", kRETENTIONARG);
00528             fprintf(stderr, errbuf);
00529             return 1;
00530         }
00531         else if (retention < 0)
00532         {
00533             snprintf(errbuf, sizeof(errbuf), "The value for %s must be a 15-bit positive integer.", kRETENTIONARG);
00534             fprintf(stderr, errbuf);
00535             return 1;
00536         }
00537         else
00538         {
00539             retention = (int16_t)(retention & 0x7FFF);
00540         }
00541     }
00542     
00543     newsuretention = INT16_MIN;
00544     if (drms_cmdparams_exists(&cmdparams, kNewSuRetention))
00545     {
00546         newsuretention = drms_cmdparams_get_int16(&cmdparams, kNewSuRetention, &status);
00547         if (status != DRMS_SUCCESS)
00548         {
00549             if (status == DRMS_ERROR_INVALIDCMDARGCONV)
00550             {
00551                 snprintf(errbuf, sizeof(errbuf), "The value for %s must be a 15-bit positive integer.", kNewSuRetention);
00552                 fprintf(stderr, errbuf);
00553             }
00554             
00555             snprintf(errbuf, sizeof(errbuf), "Invalid value for %s.", kNewSuRetention);
00556             fprintf(stderr, errbuf);
00557             return 1;
00558         }
00559         else if (newsuretention < 0)
00560         {
00561             snprintf(errbuf, sizeof(errbuf), "The value for %s must be a 15-bit positive integer.", kNewSuRetention);
00562             fprintf(stderr, errbuf);
00563             return 1;
00564         }
00565         else
00566         {
00567             newsuretention = (int16_t)(newsuretention & 0x7FFF);
00568         }
00569     }
00570     
00571   int query_mem = 512;
00572   if (cmdparams_exists (&cmdparams, kQUERYMEMARG)) {
00573     query_mem = cmdparams_get_int(&cmdparams, kQUERYMEMARG, NULL);
00574   }
00575     
00576     int dbtimeout = INT_MIN;
00577     if (drms_cmdparams_exists(&cmdparams, kDBTimeOut))
00578     {
00579         dbtimeout = drms_cmdparams_get_int(&cmdparams, kDBTimeOut, NULL);
00580     }
00581     
00582     int dbutf8clientencoding = 0;
00583     if (drms_cmdparams_exists(&cmdparams, kDBUtf8ClientEncoding))
00584     {
00585         dbutf8clientencoding = drms_cmdparams_get_int(&cmdparams, kDBUtf8ClientEncoding, NULL);
00586     }
00587 
00588   int loopconn = cmdparams_isflagset(&cmdparams, kLoopConn);
00589     
00590     int createshadows = cmdparams_isflagset(&cmdparams, kCreateShadows);
00591 
00592   /* Initialize server's own DRMS environment and connect to 
00593      DRMS database server. */
00594   if ((drms_env = drms_open(dbHostAndPort, dbuser,dbpasswd,dbname,sessionns)) == NULL)
00595   {
00596     fprintf(stderr,"Failure during server initialization.\n");
00597     return 1;
00598   }
00599     else if (verbose)
00600     {
00601         printf("Connected to database %s on host %s and port %s as user %s.\n", dbname, dbhost, dbport, dbuser);
00602     }
00603 
00604   /* Initialize global things. */
00605 
00606 
00607   /***************** Set up exit() and signal handling ********************/
00608 
00609   atexit(atexit_action);  
00610  
00611   /* Block signals INT, QUIT, TERM, and USR1. They will explicitly
00612      handled by the signal thread created below. */
00613 #ifndef DEBUG
00614   sigemptyset(&drms_env->signal_mask);
00615   sigaddset(&drms_env->signal_mask, SIGINT);
00616   sigaddset(&drms_env->signal_mask, SIGQUIT);
00617   sigaddset(&drms_env->signal_mask, SIGTERM);
00618   sigaddset(&drms_env->signal_mask, SIGUSR1);
00619   sigaddset(&drms_env->signal_mask, SIGUSR2);
00620 
00621   if( (status = pthread_sigmask(SIG_BLOCK, &drms_env->signal_mask, &drms_env->old_signal_mask)))
00622   {
00623     fprintf(stderr,"pthread_sigmask call failed with status = %d\n", status);          
00624     Exit(1);
00625   }
00626 
00627   drms_env->main_thread = pthread_self();
00628 
00629   /* Register sigblock function - whenever the dbase is accessed, certain signals
00630    * (like SIGUSR2) shouldn't interrupt such actions. */
00631   /* This is no longer an issue - there is no signal handler any more. Instead, shutdown 
00632    * is handled completely by the signal thread. If the signal thread attempts to 
00633    * access the db, but the main thread has the db lock, the signal thread will 
00634    * block until main relinquishes the lock. The signal-handler implementation required
00635    * blocking of the SIGUSR2 signal because it was asynchronously called, which could
00636    * cause interruption of a thread holding the db lock. Then the signal handler
00637    * would try to obtain the db lock, resulting in deadlock.
00638    */
00639   // db_register_sigblock(&drms_createsigmask, &drms_env->main_thread);
00640 
00641 
00642    /* Free cmd-params (valgrind reports this - let's just clean up so it doesn't show up on 
00643     * valgrind's radar). */
00644   CleanerData_t cleaner = {(pFn_Cleaner_t)FreeCmdparams, (void *)NULL};
00645   
00646   drms_server_registercleaner(drms_env, &cleaner);
00647 
00648   /* Spawn a thread that handles signals and controls server 
00649      abort or shutdown. */
00650   if( (status = pthread_create(&drms_env->signal_thread, NULL, &drms_signal_thread, 
00651                    (void *) drms_env)) )
00652   {
00653     fprintf(stderr,"Thread creation failed: %d\n", status);          
00654     Exit(1);
00655   }
00656 
00657 #endif
00658   fflush(stdout);
00659 
00660   db_handle = drms_env->session->db_handle;
00661 
00662   drms_env->archive = archive;
00663   drms_env->retention = retention; /* SUM_get() retention. */
00664   drms_env->newsuretention = newsuretention; /* New SU retention. */
00665   drms_env->query_mem = query_mem;
00666     drms_env->dbtimeout = dbtimeout;
00667     drms_env->dbutf8clientencoding = dbutf8clientencoding;
00668   drms_env->verbose = verbose;
00669   drms_env->server_wait = 0;
00670 
00671   drms_env->dbpasswd = dbpasswd;
00672   drms_env->user = user;
00673   drms_env->logfile_prefix = module_name;
00674   drms_env->dolog = dolog;
00675   drms_env->quiet = quiet;
00676 
00677   drms_env->loopconn = loopconn;
00678     drms_env->createshadows = createshadows;
00679 
00680   int abort_flag = 1;
00681 
00682   /* check for a shutdown before doing each bit of work below - these
00683    * are the only places where shutdown is allowed */
00684   if (GetShutdownState(drms_env) == kSHUTDOWN_UNINITIATED)
00685   {
00686      if (verbose)
00687        printf("Setting isolation level to SERIALIZABLE.\n");
00688      /* Set isolation level to serializable. */
00689      if ( db_isolation_level(db_handle, DB_TRANS_SERIALIZABLE) )
00690      {
00691         fprintf(stderr,"Failed to set database isolation level.\n");
00692         Exit(1);
00693      }
00694 
00695      if (GetShutdownState(drms_env) == kSHUTDOWN_UNINITIATED)
00696      {
00697         char hostname[1024];
00698         if (gethostname(hostname, sizeof(hostname))) {
00699            Exit(1);
00700         }
00701         strncpy(drms_env->session->hostname, hostname, DRMS_MAXHOSTNAME);
00702 
00703         if (GetShutdownState(drms_env) == kSHUTDOWN_UNINITIATED)
00704         {
00705            drms_server_begin_transaction(drms_env);
00706            abort_flag = CallDoIt();
00707         }
00708      }
00709   }
00710 
00711   /* Doesn't return - closes SUMS and signal threads, cleans up environment, closes dbase connection, frees cmdparams */
00712   /* If we get here, then prevent the signal thread from initiating a shutdown - main will do it instead. */
00713   FinalBits(drms_env, abort_flag);
00714 
00715   /* execution doesn't get here */
00716   return 0;
00717 }
00718
base/drms/libs/main/jsoc_main.c
