libdb/examples/csharp/excs_repquote/RepQuoteExample.cs

/*-
 * See the file LICENSE for redistribution information.
 *
 * Copyright (c) 2009, 2011 Oracle and/or its affiliates.  All rights reserved.
 *
 */
using System;
using System.Collections.Generic;
using System.IO;
using System.Text;
using System.Threading;

using BerkeleyDB;

/**
 * RepQuoteExample is a simple but complete demonstration of a replicated
 * application. The application is a mock stock ticker. The master accepts a
 * stock symbol and an numerical value as input, and stores this information
 * into a replicated database; either master or clients can display the
 * contents of the database.
 * 
 * The options to start a given replication node are:
 * 
 *   -h home (required; h stands for home directory)
 *   -l host:port (required unless -L is specified; l stands for local)
 *   -L host:port (optional, L means group creator)
 *   -C or M (optional; start up as client or master)
 *   -r host:port (optional; r stands for remote; any number of these may
 *      be specified)
 *   -R host:port (optional; R stands for remote peer; only one of these may
 *      be specified)
 *   -a all|quorum (optional; a stands for ack policy)
 *   -b (optional; b stands for bulk)
 *   -p priority (optional; defaults to 100)
 *   -v (optional; v stands for verbose)
 * 
 * A typical session begins with a command such as the following to start a
 * master:
 *
 * excs_repquote.exe -M -h dir1 -l localhost:6000
 * 
 * and several clients:
 *
 * excs_repquote.exe -C -h dir2
 *               -l localhost:6001 -r localhost:6000
 * excs_repquote.exe -C -h dir3
 *               -l localhost:6002 -r localhost:6000
 * excs_repquote.exe -C -h dir4
 *               -l localhost:6003 -r localhost:6000
 * 
 * Each process is a member of a DB replication group. The sample application
 * expects the following commands to stdin:
 * 
 * NEWLINE -- print all the stocks held in the database
 * quit -- shutdown this node
 * exit -- shutdown this node
 * stock_symbol number -- enter this stock and number into the database
 */

namespace excs_repquote
{
	public class RepQuoteExample
	{
		private RepQuoteEnvironment dbenv;
		private Thread checkpointThread;
		private Thread logArchiveThread;

		public static void usage()
		{
			Console.WriteLine(
			    "usage: " + RepConfig.progname +
			    " -h home -l|-L host:port\n" +
			    " [-CM][-r host:port][-R host:port]\n" +
			    " [-a all|quorum][-b][-p priority][-v]");

			Console.WriteLine(
			    "\t -h home (required; h stands for home directory)\n" +
			    "\t -l host:port (required unless -L is specified;" +
			    " l stands for local)\n" +
			    "\t -L host:port (optional, L means group creator)" +
			    "\t -C or -M (optional; start up as client or master)\n" +
			    "\t -r host:port (optional; r stands for remote; any number " +
			    "of these\n" +
			    "\t    may be specified)\n" +
			    "\t -R host:port (optional; R stands for remote peer; only " +
			    "one of\n" +
			    "\t    these may be specified)\n" +
			    "\t -a all|quorum (optional; a stands for ack policy)\n" +
			    "\t -b (optional; b stands for bulk)\n" +
			    "\t -p priority (optional; defaults to 100)\n" +
			    "\t -v (optional; v stands for verbose)\n");

			Environment.Exit(1);
		}

		public static void Main(string[] args)
		{
			RepConfig config = new RepConfig();
			bool isPeer, isCreator;
			uint tmpPort = 0;

            /*
             * RepQuoteExample is meant to be run from build_windows\AnyCPU, in
             * either the Debug or Release directory. The required core
             * libraries, however, are in either build_windows\Win32 or
             * build_windows\x64, depending upon the platform.  That location
             * needs to be added to the PATH environment variable for the
             * P/Invoke calls to work.
             */
            try {
                String pwd = Environment.CurrentDirectory;
                pwd = Path.Combine(pwd, "..");
                pwd = Path.Combine(pwd, "..");
                if (IntPtr.Size == 4)
                    pwd = Path.Combine(pwd, "Win32");
                else
                    pwd = Path.Combine(pwd, "x64");
#if DEBUG
                pwd = Path.Combine(pwd, "Debug");
#else
                pwd = Path.Combine(pwd, "Release");
#endif
                pwd += ";" + Environment.GetEnvironmentVariable("PATH");
                Environment.SetEnvironmentVariable("PATH", pwd);
            } catch (Exception e) {
                Console.WriteLine(
                    "Unable to set the PATH environment variable.");
                Console.WriteLine(e.Message);
                return;
            }
            
            /*  Extract the command line parameters. */
			for (int i = 0; i < args.Length; i++)
			{
				isPeer = false;
				isCreator = false;
				string s = args[i];
				if (s[0] != '-')
					continue;
				switch (s[1])
				{
					case 'a':
						if (i == args.Length - 1)
							usage();
						i++;
						if (args[i].Equals("all"))
							config.ackPolicy = AckPolicy.ALL;
						else if (!args[i].Equals("quorum"))
							usage();
						break;
					case 'b':
						config.bulk = true;
						break;
					case 'C':
						config.startPolicy = StartPolicy.CLIENT;
						break;
					case 'h':
						if (i == args.Length - 1)
							usage();
						i++;
						config.home = args[i];
						break;
					case 'l':
					case 'L':
						if (i == args.Length - 1)
							usage();
						if (args[i].Equals("-L"))
							isCreator = true;
						i++;
						string[] words = args[i].Split(':');
						if (words.Length != 2)
						{
							Console.Error.WriteLine("Invalid host " + 
							    "specification host:port needed.");
							usage();
						}
						try 
						{
							tmpPort = uint.Parse(words[1]);
						} catch (InvalidCastException)
						{
							Console.Error.WriteLine("Invalid host " + 
							    "specification, could not parse port number.");
							usage();
						}
						config.localSite.Host = words[0];
						config.localSite.Port = tmpPort;
						config.localSite.GroupCreator = isCreator;
						config.localSite.LocalSite = true;
						break;
					case 'M':
						config.startPolicy = StartPolicy.MASTER;
						break;
					case 'p':
						if (i == args.Length - 1)
							usage();
						i++;
						try 
						{
						    config.priority = uint.Parse(args[i]);
						} catch (InvalidCastException) 
						{
						    Console.Error.WriteLine("Unable to parse priority.");
						    usage();
						}
						break;
					case 'r':
					case 'R':
						if (i == args.Length - 1)
							usage();
						if (args[i].Equals("-R"))
							isPeer = true;
						i++;
						words = args[i].Split(':');
						if (words.Length != 2)
						{
							Console.Error.WriteLine("Invalid host " + 
							    "specification host:port needed.");
							usage();
						}
						try 
						{
							tmpPort = uint.Parse(words[1]);
						} catch (InvalidCastException)
						{
							Console.Error.WriteLine("Invalid host " + 
							    "specification, could not parse port number.");
							usage();
						}
						DbSiteConfig remoteConfig = new DbSiteConfig();
						remoteConfig.Helper = true;
						remoteConfig.Host = words[0];
						remoteConfig.Port = tmpPort;
						remoteConfig.Peer = isPeer;
						config.remoteSites.Add(remoteConfig);
						break;
					case 'v':
						config.verbose = true;
						break;
					default:
						Console.Error.WriteLine(
						    "Unrecognized option: " + args[i]);
						usage();	
						break;
				}
			}

			/* Error check command line. */
			if (config.localSite.Host == null || config.home.Length == 0)
				usage();

			RepQuoteExample runner = null;
			try 
			{
				runner = new RepQuoteExample();
				runner.init(config);
				runner.doloop();
				runner.terminate();
				runner = null;
			} catch (DatabaseException dbErr) 
			{
				Console.Error.WriteLine("Caught an exception during " +
				    "initialization or processing: " + dbErr);
				if (runner != null)
					runner.terminate();
			}
		} /* End main. */

		public RepQuoteExample()
		{
			dbenv = null;
		}

		public int init(RepConfig config)
		{
			int ret = 0;

			DatabaseEnvironmentConfig envConfig = new DatabaseEnvironmentConfig();
			envConfig.ErrorPrefix = RepConfig.progname;
			envConfig.RepSystemCfg = new ReplicationConfig();

			envConfig.RepSystemCfg.RepmgrSitesConfig.Add(config.localSite);
			for (int i = 0; i < config.remoteSites.Count; i++)
				envConfig.RepSystemCfg.RepmgrSitesConfig.Add(config.remoteSites[i]);

			envConfig.RepSystemCfg.BulkTransfer = config.bulk;

			/*
			 * Configure heartbeat timeouts so that repmgr monitors the
			 * health of the TCP connection.  Master sites broadcast a heartbeat
			 * at the frequency specified by the DB_REP_HEARTBEAT_SEND timeout.
			 * Client sites wait for message activity the length of the 
			 * DB_REP_HEARTBEAT_MONITOR timeout before concluding that the 
			 * connection to the master is lost.  The DB_REP_HEARTBEAT_MONITOR 
			 * timeout should be longer than the DB_REP_HEARTBEAT_SEND timeout.
			 */
			envConfig.RepSystemCfg.HeartbeatMonitor = 10000000;
			envConfig.RepSystemCfg.HeartbeatSend = 5000000;

			/* 
			 * Set replication group election priority for this environment.
			 * An election first selects the site with the most recent log
			 * records as the new master.  If multiple sites have the most
			 * recent log records, the site with the highest priority value 
			 * is selected as master.
			 */
			envConfig.RepSystemCfg.Priority = config.priority;
			envConfig.MPoolSystemCfg = new MPoolConfig();
			envConfig.MPoolSystemCfg.CacheSize = RepConfig.CACHESIZE;
			envConfig.TxnNoSync = true;

			envConfig.EventNotify = new EventNotifyDelegate(RepQuoteEventHandler);

			/*
			 * Set the policy that determines how master and client sites
			 * handle acknowledgement of replication messages needed for
			 * permanent records.  The default policy of "quorum" requires only
			 * a quorum of electable peers sufficient to ensure a permanent
			 * record remains durable if an election is held.  The "all" option
			 * requires all clients to acknowledge a permanent replication
			 * message instead.
			 */
			envConfig.RepSystemCfg.RepMgrAckPolicy = config.ackPolicy;

			/*
			 * Set the threshold for the minimum and maximum time the client
			 * waits before requesting retransmission of a missing message.
			 * Base these values on the performance and load characteristics
			 * of the master and client host platforms as well as the round
			 * trip message time.
			 */
			envConfig.RepSystemCfg.RetransmissionRequest(20000, 500000);

			/*
			 * Configure deadlock detection to ensure that any deadlocks
			 * are broken by having one of the conflicting lock requests
			 * rejected. DB_LOCK_DEFAULT uses the lock policy specified
			 * at environment creation time or DB_LOCK_RANDOM if none was
			 * specified.
			 */
			envConfig.LockSystemCfg = new LockingConfig();
			envConfig.LockSystemCfg.DeadlockResolution = DeadlockPolicy.DEFAULT;

			envConfig.Create = true;
			envConfig.RunRecovery = true;
			envConfig.FreeThreaded = true;
			envConfig.UseReplication = true;
			envConfig.UseLocking = true;
			envConfig.UseLogging = true;
			envConfig.UseMPool = true;
			envConfig.UseTxns = true;
			envConfig.Verbosity = new VerboseMessages();
			envConfig.Verbosity.Replication = config.verbose;

			try 
			{
				dbenv = RepQuoteEnvironment.Open(config.home, envConfig);
			} catch(DatabaseException e) 
			{
				Console.WriteLine("Fail to open environment: " + e.Message);
				return 1;
			}

			
			/* The following base replication features may also be useful to your
			 * application. See Berkeley DB documentation for more details.
			 *   - Master leases: Provide stricter consistency for data reads
			 *     on a master site.
			 *   - Timeouts: Customize the amount of time Berkeley DB waits
			 *     for such things as an election to be concluded or a master
			 *     lease to be granted.
			 *   - Delayed client synchronization: Manage the master site's
			 *     resources by spreading out resource-intensive client 
			 *     synchronizations.
			 *   - Blocked client operations: Return immediately with an error
			 *     instead of waiting indefinitely if a client operation is
			 *     blocked by an ongoing client synchronization.
			 *
			 * The following repmgr features may also be useful to your
			 * application.  See Berkeley DB documentation for more details.
			 *  - Two-site strict majority rule - In a two-site replication
			 *    group, require both sites to be available to elect a new
			 *    master.
			 *  - Timeouts - Customize the amount of time repmgr waits
			 *    for such things as waiting for acknowledgements or attempting 
			 *    to reconnect to other sites.
			 *  - Site list - return a list of sites currently known to repmgr.
			 */

			/* Start checkpoint and log archive support threads. */ 
			checkpointThread = new Thread(new ThreadStart(CheckPoint));
			checkpointThread.Start();
			logArchiveThread = new Thread(new ThreadStart(LogArchive));
			logArchiveThread.Start();

			/* Start replication manager. */
			if (config.startPolicy == StartPolicy.CLIENT)
				dbenv.env.RepMgrStartClient(3);
			else if (config.startPolicy == StartPolicy.ELECTION)
				dbenv.env.RepMgrStartClient(3, true);
			else if (config.startPolicy == StartPolicy.MASTER)
				dbenv.env.RepMgrStartMaster(3);

			return ret;
		}

		public int doloop()
		{
			BTreeDatabase db = null;
			    
			for (;;)
			{
				if (db == null) 
				{
					BTreeDatabaseConfig dbConfig = new BTreeDatabaseConfig();
					dbConfig.Env = dbenv.env;

					if (dbenv.IsMaster) 
					{
						/*
						 * Open database allowing create only if this is a master
						 * database.  A client database uses polling to attempt
						 * to open the database, without creating it, until the
						 * open succeeds.
						 *
						 * This polling logic for allowing create can be 
						 * simplified under some circumstances.  For example, if
						 * the application can be sure a database is already
						 * there, it would never need to open it allowing create.
						 */
							dbConfig.Creation = CreatePolicy.IF_NEEDED;
					}

					dbConfig.AutoCommit = true;

					try 
					{
						db = BTreeDatabase.Open(RepConfig.progname, dbConfig);
					} catch (DatabaseException) 
					{
						Console.WriteLine("no stock database available yet.");
						if (db != null) 
						{
							db.Close(true);
							db = null;
						}

						Thread.Sleep(RepConfig.SLEEPTIME);
						continue;
					}
				}

				/* Listen for input, and add it to the database. */
				Console.Write("QUOTESERVER");
				if (dbenv.IsMaster == false)
					Console.Write("(read-only)");
				Console.Write("> ");
				string nextLine = null;
				try
				{
					nextLine = Console.ReadLine();
				} catch(System.IO.IOException)
				{
					Console.WriteLine("Unable to get data");
					break;
				}

				 /* A blank line causes the DB to be dumped. */
				string[] words = nextLine.Split(' ');
				if (words.Length == 0 ||
				    words.Length == 1 && words[0].Length == 0) 
				{
					try
					{
						if (dbenv.InClientSync)
							Console.WriteLine("Cannot read data during " +
							    "client initialization - please try again.");
						else
							printStocks(db);
					} catch (DeadlockException) 
					{
						continue;
					} catch (DatabaseException e) 
					{
						/*
						 * This could be DB_REP_HANDLE_DEAD, which
						 * should close the database and continue.
						 */
						Console.WriteLine("Got db exception reading replication"
						    + "DB: " + e);
						Console.WriteLine("Expected if it was due to a dead " +
						    "replication handle, otherwise an unexpected error.");
						db.Close(false);/* Close no sync. */
						db = null;
						continue;
					}
					continue;
				}

				if (words.Length == 1 &&
				    (words[0].ToLower().Equals("quit") || 
				    words[0].ToLower().Equals("exit"))) 
				{
					dbenv.AppFinished = true;
					break;
				} else if (words.Length != 2) 
				{
					Console.WriteLine("Format: TICKER VALUE");
					continue;
				}

				if (!dbenv.IsMaster)
				{
					Console.WriteLine("Can't update client");
					continue;
				}

				DatabaseEntry key = new DatabaseEntry(
				    ASCIIEncoding.ASCII.GetBytes(words[0]));
				DatabaseEntry data = new DatabaseEntry(
				    ASCIIEncoding.ASCII.GetBytes(words[1]));

				db.Put(key, data);
			}

			if (db != null)
				db.Close(true);

			return 0;
		}

		public void terminate()
		{
			/* Wait for checkpoint and log archive threads to finish. */
			try 
			{
				logArchiveThread.Join();
				checkpointThread.Join();
			} catch (Exception) 
			{
				Console.WriteLine("Support thread join failed.");
			}

			/*
			 * We have used the DB_TXN_NOSYNC environment flag for improved
			 * performance without the usual sacrifice of transactional durability,
			 * as discussed in the "Transactional guarantees" page of the Reference
			 * Guide: if one replication site crashes, we can expect the data to
			 * exist at another site.  However, in case we shut down all sites
			 * gracefully, we push out the end of the log here so that the most
			 * recent transactions don't mysteriously disappear.
			 */
			dbenv.env.LogFlush();
			dbenv.env.Close();
		}

		/*
		 * void return type since error conditions are propogated
		 * via exceptions.
		 */
		private void printStocks(Database db)
		{
			Cursor dbc = db.Cursor();
			Console.WriteLine("\tSymbol\t\tPrice");
			Console.WriteLine("\t======\t\t=====");
			DatabaseEntry key = new DatabaseEntry();
			DatabaseEntry data = new DatabaseEntry();

			foreach (KeyValuePair<DatabaseEntry, DatabaseEntry> pair in dbc)
			{
				string keyStr = ASCIIEncoding.ASCII.GetString(pair.Key.Data);
				string dataStr = ASCIIEncoding.ASCII.GetString(pair.Value.Data);
				Console.WriteLine("\t"+keyStr+"\t\t"+dataStr);
			}
			dbc.Close();
		}

		/*
		 * This is a very simple thread that performs checkpoints at a fixed
		 * time interval.  For a master site, the time interval is one minute
		 * plus the duration of the checkpoint_delay timeout (30 seconds by
		 * default.)  For a client site, the time interval is one minute.
		 */
		public void CheckPoint()
		{
			for (; ; )
			{
				/*
				 * Wait for one minute, polling once per second to see if
				 * application has finished.  When application has finished,
				 * terminate this thread.
				 */
				for (int i = 0; i < 60; i++)
				{
					Thread.Sleep(1000);
					if (dbenv.AppFinished)
						return;
				}

				/* Perform a checkpoint. */
				try
				{
					dbenv.env.Checkpoint();
				} catch (DatabaseException)
				{
					Console.WriteLine("Could not perform checkpoint.");
				}
			}
		}

		/*
		 * This is a simple log archive thread.  Once per minute, it removes all but
		 * the most recent 3 logs that are safe to remove according to a call to
		 * DBENV->log_archive().
		 *
		 * Log cleanup is needed to conserve disk space, but aggressive log cleanup
		 * can cause more frequent client initializations if a client lags too far
		 * behind the current master.  This can happen in the event of a slow client,
		 * a network partition, or a new master that has not kept as many logs as the
		 * previous master.
		 *
		 * The approach in this routine balances the need to mitigate against a
		 * lagging client by keeping a few more of the most recent unneeded logs
		 * with the need to conserve disk space by regularly cleaning up log files.
		 * Use of automatic log removal (DBENV->log_set_config() DB_LOG_AUTO_REMOVE 
		 * flag) is not recommended for replication due to the risk of frequent 
		 * client initializations.
		 */
		public void LogArchive()
		{
			int logKeep = 3;
			int minLog;
			List<string> logFileList;

			for (; ; )
			{
				/*
				 * Wait for one minute, polling once per second to see if
				 * application has finished.  When application has finished,
				 * terminate this thread.
				 */
				for (int i = 0; i < 60; i++)
				{
					Thread.Sleep(1000);
					if (dbenv.AppFinished)
						return;
				}

				try
				{
					/* Get the list of unneeded log files. */
					logFileList = dbenv.env.ArchivableLogFiles(true);
					/* 
					 * Remove all but the logs_to_keep most recent unneeded
					 * log files. 
					 */
					minLog = logFileList.Count - logKeep;
					for (int i = 0; i < minLog; i++)
					{
						FileInfo logFile = new FileInfo(logFileList[i]);
						logFile.Delete();
					}
				} catch (DatabaseException)
				{
					Console.WriteLine("Problem deleting log archive files.");
				}
			}
		}

		
		/*
		 * Implemention of EventHandler interface to handle the Berkeley DB events
		 * we are interested in receiving.
		 */
		private void RepQuoteEventHandler(NotificationEvent eventCode, byte[] event_info)
		{
			switch (eventCode) 
			{
				case NotificationEvent.REP_CLIENT:
					dbenv.IsMaster = false;
					dbenv.InClientSync = true;
					break;
				case NotificationEvent.REP_MASTER:
					dbenv.IsMaster = true;
					dbenv.InClientSync = false;
					break;
				case NotificationEvent.REP_NEWMASTER:
					dbenv.InClientSync = true;
					break;
				case NotificationEvent.REP_STARTUPDONE:
					dbenv.InClientSync = false;
					break;
				/*
				 * Did not get enough acks to guarantee transaction
				 * durability based on the configured ack policy.  This 
				 * transaction will be flushed to the master site's
				 * local disk storage for durability.
				 */ 
				case NotificationEvent.REP_PERM_FAILED:
					Console.WriteLine("Insufficient acknowledgements " +
					    "to guarantee transaction durability.");
					break;
				default:
					Console.WriteLine("Ignoring event: {0}", eventCode);
					break;
			}
		}
	}
}