scheduler.cs

SharpNuke源代码

using System;
using System.Data;
using System.Collections;
using System.Threading;
using System.Xml;

using DotNetNuke;
using DotNetNuke.Common;
using DotNetNuke.Common.Utilities;
using DotNetNuke.UI.Skins;
using DotNetNuke.Services.Exceptions;
using DotNetNuke.Services.Log.EventLog;

//
// DotNetNuke -  http://www.dotnetnuke.com
// Copyright (c) 2002-2005
// by Shaun Walker ( [e-mail] ) of Perpetual Motion Interactive Systems Inc. ( http://www.perpetualmotion.ca )
//
// Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated
// documentation files (the "Software"), to deal in the Software without restriction, including without limitation
// the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and
// to permit persons to whom the Software is furnished to do so, subject to the following conditions:
//
// The above copyright notice and this permission notice shall be included in all copies or substantial portions
// of the Software.
//
// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED
// TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL
// THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF
// CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
// DEALINGS IN THE SOFTWARE.
//

namespace DotNetNuke.Services.Scheduling.DNNScheduling
{
	sealed class Scheduler
	{
		public class CoreScheduler
		{
			///''''''''''''''''''''''''''''''''''''''''''''''''
			//This is the heart of the scheduler mechanism.
			//This class manages running new events according
			//to the schedule.
			//
			//This class can also react to the three
			//scheduler events (Started, Progressing and Completed)
			///''''''''''''''''''''''''''''''''''''''''''''''''
			
			private static bool threadPoolInitialized = false;
			///''''''''''''''''''''''''''''''''''''''''''''''''
			//The MaxThreadCount establishes the maximum
			//threads you want running simultaneously
			//for spawning SchedulerClient processes
			///''''''''''''''''''''''''''''''''''''''''''''''''
			private static int maxThreadCount;
			private static int activeThreadCount;
			///''''''''''''''''''''''''''''''''''''''''''''''''
			//If KeepRunning gets switched to false,
			//the scheduler stops running.
			///''''''''''''''''''''''''''''''''''''''''''''''''
			private static bool forceReloadSchedule = false;
			private static bool debug = false;
			
			private static int numberOfProcessGroups;
			private static ArrayList scheduleQueue;
			private static ArrayList scheduleInProgress;
			
			///''''''''''''''''''''''''''''''''''''''''''''''''
			//The ScheduleQueue collection contains the current
			//queue of scheduler clients that need to run
			///''''''''''''''''''''''''''''''''''''''''''''''''
			private static ArrayList ScheduleQueue
			{
				get
				{
					try
					{
						queueReadWriteLock.AcquireReaderLock(readTimeout);
						try
						{
							// It is safe for this thread to read or write
							// from the shared resource.
							if (scheduleQueue == null)
							{
								scheduleQueue = new ArrayList();
							}
							Interlocked.Increment(ref reads);
						}
						finally
						{
							// Ensure that the lock is released.
							queueReadWriteLock.ReleaseReaderLock();
						}
					}
					catch (ApplicationException exception)
					{
						// The writer lock request timed out.
						Interlocked.Increment(ref readTimeout);
						Exceptions.Exceptions.LogException(exception);
					}
					return scheduleQueue;
				}
				set
				{
					try
					{
						queueReadWriteLock.AcquireWriterLock(writeTimeout);
						try
						{
							// It is safe for this thread to read or write
							// from the shared resource.
							DataCache.SetCache("ScheduleQueue", value);
							scheduleQueue = value;
							Interlocked.Increment(ref writes);
						}
						finally
						{
							// Ensure that the lock is released.
							queueReadWriteLock.ReleaseWriterLock();
						}
					}
					catch (ApplicationException exception)
					{
						// The writer lock request timed out.
						Interlocked.Increment(ref writerTimeouts);
						Exceptions.Exceptions.LogException(exception);
					}
				}
			}
			
			///''''''''''''''''''''''''''''''''''''''''''''''''
			//The ScheduleInProgress collection contains the
			//collection of tasks that are currently in progress
			///''''''''''''''''''''''''''''''''''''''''''''''''
			private static ArrayList ScheduleInProgress
			{
				get
				{
					try
					{
						queueReadWriteLock.AcquireReaderLock(readTimeout);
						try
						{
							// It is safe for this thread to read or write
							// from the shared resource.
							if (scheduleInProgress == null)
							{
								scheduleInProgress = new ArrayList();
							}
							Interlocked.Increment(ref reads);
						}
						finally
						{
							// Ensure that the lock is released.
							queueReadWriteLock.ReleaseReaderLock();
						}
					}
					catch (ApplicationException exception)
					{
						// The writer lock request timed out.
						Interlocked.Increment(ref readTimeout);
						Exceptions.Exceptions.LogException(exception);
					}
					return scheduleInProgress;
				}
				set
				{
					try
					{
						queueReadWriteLock.AcquireWriterLock(writeTimeout);
						try
						{
							// It is safe for this thread to read or write
							// from the shared resource.
							DataCache.SetCache("ScheduleQueue", value);
							scheduleInProgress = value;
							Interlocked.Increment(ref writes);
						}
						finally
						{
							// Ensure that the lock is released.
							queueReadWriteLock.ReleaseWriterLock();
						}
					}
					catch (ApplicationException exception)
					{
						// The writer lock request timed out.
						Interlocked.Increment(ref writerTimeouts);
						Exceptions.Exceptions.LogException(exception);
					}
				}
			}
			
			
			///''''''''''''''''''''''''''''''''''''''''''''''''
			//This is our array that holds the process group
			//where our threads will be kicked off.
			///''''''''''''''''''''''''''''''''''''''''''''''''
			private static ProcessGroup[] arrProcessGroup;
			
			///''''''''''''''''''''''''''''''''''''''''''''''''
			//A ReaderWriterLock will protect our objects
			//in memory from being corrupted by simultaneous
			//thread operations.  This block of code below
			//establishes variables to help keep track
			//of the ReaderWriter locks.
			///''''''''''''''''''''''''''''''''''''''''''''''''
			private static ReaderWriterLock inProgressReadWriteLock = new ReaderWriterLock();
			private static ReaderWriterLock queueReadWriteLock = new ReaderWriterLock();
			private static int readerTimeouts = 0;
			private static int writerTimeouts = 0;
			private static int reads = 0;
			private static int writes = 0;
			private static int readTimeout = 45000; //wait 45 seconds
			private static int writeTimeout = 45000; //wait 45 seconds
			private static ReaderWriterLock statusReadWriteLock = new ReaderWriterLock();
			private static DotNetNuke.Services.Scheduling.ScheduleStatus status = ScheduleStatus.STOPPED;
			
			public static bool KeepThreadAlive = true;
			public static bool KeepRunning = true;
			
			///''''''''''''''''''''''''''''''''''''''''''''''''
			//FreeThreads tracks how many threads we have
			//free to work with at any given time.
			///''''''''''''''''''''''''''''''''''''''''''''''''
			public static int FreeThreads
			{
				get { return maxThreadCount - activeThreadCount; }
			}
			
			public static int GetActiveThreadCount()
			{
				return activeThreadCount;
			}
			
			public static int GetFreeThreadCount()
			{
				return FreeThreads;
			}
			
			public static int GetMaxThreadCount()
			{
				return maxThreadCount;
			}
			
			public CoreScheduler(int maxThreads) 
			{
				if (! threadPoolInitialized)
				{
					InitializeThreadPool(maxThreads);
				}
			}
			
			public CoreScheduler(bool boolDebug, int maxThreads) 
			{
				debug = boolDebug;
				if (! threadPoolInitialized)
				{
					InitializeThreadPool(maxThreads);
				}
			}
			
			private void InitializeThreadPool (int maxThreads)
			{
				if (maxThreads == - 1)
				{
					maxThreads = 1;
				}
				numberOfProcessGroups = maxThreads;
				arrProcessGroup = new ProcessGroup[numberOfProcessGroups];
				maxThreadCount = maxThreads;
				for (int i = 0; i < numberOfProcessGroups; i++)
				{
					arrProcessGroup[i] = new ProcessGroup();
				}
				threadPoolInitialized = true;
			}
			
			public static void ReloadSchedule ()
			{
				forceReloadSchedule = true;
			}
			
			static public ScheduleStatus GetScheduleStatus()
			{
				ScheduleStatus scheduleStatus = ScheduleStatus.NOT_SET;
				try
				{
					statusReadWriteLock.AcquireReaderLock(readTimeout);
					try
					{
						// It is safe for this thread to read from
						// the shared resource.
						scheduleStatus = status;
					}
					finally
					{
						// Ensure that the lock is released.
						statusReadWriteLock.ReleaseReaderLock();
					}
				}
				catch (ApplicationException)
				{
					// The reader lock request timed out.
					Interlocked.Increment(ref readerTimeouts);
				}
				return scheduleStatus;
			}
			
			public static void SetScheduleStatus (ScheduleStatus scheduleStatus)
			{
				try
				{
					statusReadWriteLock.AcquireWriterLock(writeTimeout);
					try
					{
						// It is safe for this thread to read or write
						// from the shared resource.
						status = scheduleStatus;
						Interlocked.Increment(ref writes);
					}
					finally
					{
						// Ensure that the lock is released.
						statusReadWriteLock.ReleaseWriterLock();
					}
				}
				catch (ApplicationException exception)
				{
					// The writer lock request timed out.
					Interlocked.Increment(ref writerTimeouts);
					Exceptions.Exceptions.LogException(exception);
				}
			}
			
			public static void Halt (string sourceOfHalt)
			{
				if (Services.Scheduling.SchedulingProvider.SchedulerMode != SchedulerMode.REQUEST_METHOD)
				{
					EventLogController eventLog = new EventLogController();
					LogInfo eventLogInfo = new LogInfo();
					SetScheduleStatus(ScheduleStatus.SHUTTING_DOWN);
					eventLogInfo = new LogInfo();
					eventLogInfo.AddProperty("Initiator", sourceOfHalt);
					eventLogInfo.LogTypeKey = "SCHEDULER_SHUTTING_DOWN";
					eventLog.AddLog(eventLogInfo);
					
					KeepRunning = false;
					
					//wait for up to 120 seconds for thread
					//to shut down
					for (int i = 0; i <= 120; i++)
					{
						if (GetScheduleStatus() == ScheduleStatus.STOPPED)
						{
							return;
						}
						Thread.Sleep(1000);
					}
				}
				activeThreadCount = 0;
			}
			
			///''''''''''''''''''''''''''''''''''''''''''''''''
			//This is a multi-thread safe method that adds
			//an item to the collection of schedule items in
			//progress.  It first obtains a write lock
			//on the ScheduleInProgress object.
			///''''''''''''''''''''''''''''''''''''''''''''''''
			private static void AddToScheduleInProgress(ScheduleHistoryItem scheduleHistoryItem)
			{
				try
				{
					inProgressReadWriteLock.AcquireWriterLock(writeTimeout);
					try
					{
						// It is safe for this thread to read or write
						// from the shared resource.
						ScheduleInProgress.Add(scheduleHistoryItem);//, scheduleHistoryItem.ScheduleID.ToString());
						Interlocked.Increment(ref writes);
					}
					finally
					{
						// Ensure that the lock is released.
						inProgressReadWriteLock.ReleaseWriterLock();
					}
				}
				catch (ApplicationException exception)
				{
					// The writer lock request timed out.
					Interlocked.Increment(ref writerTimeouts);
					Exceptions.Exceptions.LogException(exception);
				}