}

/// <summary>

/// вызван метод Commit

/// </summary>

public event TransactionEventHandler OnCommit;

/// <summary>

/// вызыван метод Rollback

/// </summary>

public event TransactionEventHandler OnRollback;

/// <summary>

/// транзакция завершена (независимо от результата)

/// </summary>

public event TransactionEventHandler Complete;

/// <summary>

/// откат транзакции

/// </summary>

public void Rollback()

{

if (CompleteHandle.WaitOne(0))

return; //Этот хэндл выставляется когда у транзакции уже очистились ресурсы, поэтому в продолжении нет смысла

lock (_locker)

{

if (_commitFlag)

return;

//throw new Exception("Не могу сделать Rollback, когда уже был вызван Commit");

if (_rollbackFlag && (WorkingThread.Current == null || !WorkingThread.Current.Equals(_myThread)))

return; //Попытка повторно поставить в очередь откат

_rollbackFlag = true;

_myThread.OnIdle -= _myThread_OnIdle;

}

if (WorkingThread.Current != null && WorkingThread.Current.Equals(_myThread))

{

if (OnRollback != null)

OnRollback(this, null);

_cTrans.Rollback();

//добавляем задачу очистки ресурсов транзакции

_myThread.queue.Push(new WorkItem(new WorkItem.anywork(this.FreeResources)));

}

else

{

WorkItem.anywork w = () => { this.Rollback(); };

WorkItem wi = new WorkItem(w, null);

EnqueueWorkItem(wi);

wi.AsyncWaitHandle.WaitOne();

//_completeHandle.WaitOne();

}

}

/// <summary>

/// накат транзакции

/// </summary>

/// <remarks> если метод вызван из контекста транзакции или рабочего потока,

/// то в очередь транзакции помещается задание завершения транзакции и работа метода завершается

/// Если метод вызван вне контекста рабочих потоков,

/// то выполняется ожидание фактического завершения транзакции

/// </remarks>

public void Commit()

{

if (CompleteHandle.WaitOne(0))

return; //Этот хэндл выставляется когда у транзакции уже очистились ресурсы, поэтому в продолжении нет смысла

//сделать чтобы коммит или роллбэк был только один

lock (_locker)

{

if (_lastException != null) //уже было какое-то исключение

throw new InvalidOperationException("При выполнении транзакции возникло исключение", _lastException);

if (_rollbackFlag)

{

return;

//throw new Exception("Не могу вызвать Commit, так как уже был вызван Rollback");

}

if (_commitFlag && (WorkingThread.Current == null || !WorkingThread.Current.Equals(_myThread)))

return; //Попытка повторно поставить в очередь накат

_commitFlag = true;

_myThread.OnIdle -= _myThread_OnIdle;

}

if (WorkingThread.Current != null && WorkingThread.Current.Equals(_myThread))

{

if (OnCommit != null)

OnCommit(this, null);

_connection.Commit();

_cTrans.Commit();

//Применяем изменения к файлам транзакции

//Перед этим должны быть созданы по идее резервные копии..

foreach (TransactionFile tfile in _transactionFiles.Values)

tfile.Flush();

//добавляем задачу очистки ресурсов транзакции

_myThread.queue.Push(new WorkItem(new WorkItem.anywork(this.FreeResources)));

}

else

{

WorkItem wi = new WorkItem(new WorkItem.anywork(this.Commit));

EnqueueWorkItem(wi);

wi.AsyncWaitHandle.WaitOne();

// _completeHandle.WaitOne();

}

}

/// <summary>

/// добавляет задачу для выполнения в рамках транзакции

/// </summary>

/// <param name="wi">задача для выполнения</param>

/// <remarks>

/// Задача помещается в очередь задач потока и одновременно в очередь незавершенных задач

/// транзакции. После выполнения задачи она удаляется из очереди незавершенных задач транзакции.

///

/// При выполнении задачи в контекст устанавливается текущая транзакция

///

/// Если при выполнении задачи в транзакции возникло исключение - транзакция автоматически откатывается

/// </remarks>

/// <exception cref="InvalidOperationException">Транзакция находится в состоянии завершения и не может принимать новых задач</exception>

public void AddAction(Cognitive.Threading.WorkItem wi)

{

lock (_locker)

{

if (_rollbackFlag || _commitFlag)

throw new InvalidOperationException("Транзакция находится в состоянии завершения и не может принимать новых задач");

EnqueueWorkItem(wi);

}

}

private void Run(Cognitive.Threading.WorkItem wi)

{

lock (_locker)

{

if (_myThread.State == WorkingThreadStates.Stopping)

{ //Фиктивное завершение

(wi.AsyncWaitHandle as System.Threading.ManualResetEvent).Set();

return;

}

wi.WorkException += new WorkItemExceptionDelegate(wi_WorkException);

wi.WorkComplete += new WorkItemEventDelegate(wi_WorkComplete);

wi.WorkActivating += new WorkItemEventDelegate(wi_WorkActivating);

_running = wi;

_myThread.queue.Push(_running);

}

}

private void EnqueueWorkItem(Cognitive.Threading.WorkItem wi)

{

if (wi == null)

throw new NullReferenceException("wi не может быть null");

lock (_locker)

{

if (_running == null)

Run(wi); //ничего не выполняется - сразу запускаем

else

{

if (_myThread.State == WorkingThreadStates.Stopping)

{ //Фиктивное завершение

(wi.AsyncWaitHandle as System.Threading.ManualResetEvent).Set();

return;

}

_transactionQueue.Enqueue(wi); //кладем в очередь

}

}

}

class CWIAsync : IAsyncResult

{

WorkItem _wi;

bool _Complete = false;

public CWIAsync(WorkItem wi)

{

_wi = wi;

_wi.WorkComplete += new WorkItemEventDelegate(_wi_WorkComplete);

}

void _wi_WorkComplete(WorkItem source)

{

_Complete = true;

}

#region IAsyncResult Members

public object AsyncState

{

get { return null; }

}

public System.Threading.WaitHandle AsyncWaitHandle

{

get {

return _wi.AsyncWaitHandle;

}

}

public bool CompletedSynchronously

{

get { return _Complete; }

}

public bool IsCompleted

{

get { return _Complete; }

}

#endregion

}

public Object Invoke(Delegate del, params object[] p)

{

WorkItem wi = null;

switch (p.Length)

{

case 0:

wi = new WorkItem(del);

break;

case 1:

wi = new WorkItem(del, p[0]);

break;

case 2:

wi = new WorkItem(del, p[0], p[1]);

break;

case 3:

wi = new WorkItem(del, p[0], p[1], p[2]);

break;

case 4:

wi = new WorkItem(del, p[0], p[1], p[2], p[3]);

break;

default:

throw new NotImplementedException();

}

Exception ex = null;

WorkItemExceptionDelegate we = (s, e) => { ex = e; };

wi.WorkException += we;

/**/

AddAction(wi);

/*EnqueueWorkItem(wi);*/

wi.AsyncWaitHandle.WaitOne();

if (ex != null)

throw new System.Reflection.TargetInvocationException(ex);

return wi.Result;

}

public IAsyncResult InvokeAsync(Delegate del, params object[] p)

{

WorkItem wi = null;

switch (p.Length)

{

case 0:

wi = new WorkItem(del);

break;

case 1:

wi = new WorkItem(del, p[0]);

break;

case 2:

wi = new WorkItem(del, p[0], p[1]);

break;

case 3:

wi = new WorkItem(del, p[0], p[1], p[2]);

break;

case 4:

wi = new WorkItem(del, p[0], p[1], p[2], p[3]);

break;

default:

throw new NotImplementedException();

}

IAsyncResult ar = new CWIAsync(wi);

/**/

AddAction(wi);

/*EnqueueWorkItem(wi);*/

return ar;

}

private Microsoft.Samples.DbConnectionScope.DbConnectionScope _dbScope;

/// <summary>

/// обрабатываемый WorkItem

/// </summary>

private WorkItem CurrentWorkItem

{

get{

lock (_locker)

{

return _running;

}

}

}

/// <summary>

/// попытка войти в режим блокировки

/// </summary>

/// <param name="lockId">идентификатор блокировки</param>

/// <returns></returns>

public bool TryEnterLock(String lockId)

{

if (!this.Equals(Transaction.Current))

throw new InvalidOperationException("TryEnterLock может быть вызван только из потока транзакции");

if (Transaction.Current._server.blockMan.TryEnterBlock(lockId))

return true;

return false;

}

public bool IsInLock(String lockId)

{

return _server.blockMan.IsInLock(this, lockId);

}

/// <summary>

/// вход в режим блокировки с ожиданием

/// </summary>

/// <param name="lockId">идентификатор блокировки</param>

public void EnterLock(String lockId)

{

if (Transaction.Current != null)

throw new InvalidOperationException("Для внесения блокировки заданием следует использовать TryEnterLock вместо EnterLock");

AddAction(new CLockWorkItem(this, lockId));

}

/// <summary>

/// При выполнении задачи в контекст ее потока устанавливается текущая транзакция

/// </summary>

/// <param name="source"></param>

void wi_WorkActivating(WorkItem source)

{

lock (_locker)

{

if (!_myThread.Context.ContainsKey("Transaction"))

_myThread.Context.Add("Transaction", this);

}

///Устанавливаем текущую транзакцию

System.Transactions.Transaction.Current = _cTrans;

_dbScope = new Microsoft.Samples.DbConnectionScope.DbConnectionScope();

_dbScope.AddConnection(_sqlConn.ConnectionString, _sqlConn);

}

/// <summary>

/// Выполнение задачи завершилось и она удаляется из очереди незавершенных задач транзакции

/// </summary>

/// <param name="source"></param>

void wi_WorkComplete(WorkItem source)

{

lock (_locker)

{

WorkItem wi = _running;

//закрываем открытый scope соединения

if (_dbScope != null)

{

_dbScope.AddConnection(_sqlConn.ConnectionString, new SqlConnection());

_dbScope.Dispose();

_dbScope = null;

}

_MyIdleStart = DateTime.Now;

_running = null;

if (source is CLockWorkItem)

{

if (true != (bool)source.Result)

{

Run(((CLockWorkItem)source).Copy());

return;

}

}

///вытаскиваем следующий элемент из очереди

if (_transactionQueue.Count > 0)

Run(_transactionQueue.Dequeue());

}

}

private Exception _lastException;

/// <summary>

/// Откат, если при выполнении задачи в транзакции возникло исключение

/// </summary>

/// <param name="source"></param>

/// <param name="e"></param>

void wi_WorkException(WorkItem source, Exception e)

{

Trace.WriteLine("исключение в транзакции " + this._MyIndex.ToString());

lock(_locker)

{

//Исключение не в коммите и не в роллбэке

if (!_commitFlag && !_rollbackFlag && _lastException == null)

_lastException = e;

_commitFlag = false; //Нельзя коммитить то, что ошиблось

if (_rollbackFlag)

throw e; //падение при роллбэке обработать не можем

}

//Вызов происходит из рабочего потока, в его очередь ставится откат и, затем, освобождение

//ресурсов, которые сразу и выполняются

Rollback();

if (_lastException != null)

return;

//После этого можно кинуть исключение

throw e;

}

/// <summary>

/// создание файла в транзакции

/// </summary>

public TransactionFile CreateFile(string name)

{

TransactionFile tf = new TransactionFile(name);

lock (_locker)

{

_transactionFiles.Add(tf.id, tf);

}

return tf;

}

/// <summary>

/// возвращает созданный файл по его идентификатору

/// </summary>

public TransactionFile GetFile(string fileid)

{

lock (_locker)

{

if (_transactionFiles.ContainsKey(fileid))

return _transactionFiles[fileid];

return null;

}

}

/// <summary>

/// возврашает текущую транзакцию

/// </summary>

/// <remarks>

/// возвращает текущую транзакцию из контекста выполнения

/// </remarks>

public static Transaction Current

{

get

{

try

{

if (WorkingThread.Current == null)

return null;

return WorkingThread.Current.Context["Transaction"] as Transaction;

}

catch(NullReferenceException )

{

return null;

}

catch(KeyNotFoundException )

{

return null;

}

}

}

#region IDFObjectsProvider Members

/// <summary>

/// выполняет конструирование объектов по идентификатору в контексте транзакции

/// </summary>

/// <param name="id">полный идентификатор объекта</param>

/// <returns>созданный объект</returns>

/// <remarks>

/// Этот метод выполняет конструирование и кэширование ранее запрошенных объектов в рамках транзакции

/// При завершении транзакции выполняется полная очистка кэша

/// </remarks>

public Environment.IDFObject CreateObject(string id)

{

Environment.IDFContainerObject root = _server.CreateRootObject(this._connection);

if (String.IsNullOrEmpty(id))

return root;

///разбор пути, конструирование и кэширование объектов

String[] ObjectKeys = id.Split(new char[] { '/' }, StringSplitOptions.RemoveEmptyEntries);

Environment.IDFContainerObject tempContObj = root;

Environment.IDFObject tempObj = null;

foreach (String CurrentKey in ObjectKeys)

{

tempObj = tempContObj.GetChild(CurrentKey);

if (!(tempObj is Environment.IDFContainerObject))

return tempObj;

tempContObj = (tempObj as Environment.IDFContainerObject);

}

return tempObj;

}

#endregion

/// <summary>

/// возвращает соединение, связанное с транзакцией

/// </summary>

public CTNKDESMLib.ICTDEDatabase Connection

{

get{

return _connection;

}

}

/// <summary>

/// возвращает объект для управления взаимодействией в рамках транзакции

/// </summary>

public Interaction.Interaction Interaction

{

get

{

return new Interaction.Interaction(_connection);

}

}

/// <summary>

/// добавление объекта в конекст транзакции

/// </summary>

/// <param name="key"></param>

/// <param name="p"></param>

public void AddContextObject(String key, Object p)

{

lock (_locker)

{

if (_ctx.ContainsKey(key))

_ctx[key] = p;

else

_ctx.Add(key, p);

}

}

/// <summary>

/// получение объекта из контекста транзакции

/// </summary>

/// <param name="key"></param>

/// <returns></returns>

public Object GetContextObject(String key)

{

lock (_locker)

{

if (_ctx.ContainsKey(key))

return _ctx[key];

else return null;

}

}

}

/// <summary>

/// контекст транзакции

/// </summary>

public class TransactionContext : System.Collections.Generic.Dictionary<string, Object>

{

}

internal class CLockWorkItem : WorkItem

{

private delegate bool TryEnterLockDelegate(String x);

private String _LockId;

private Transaction _Trans;

public CLockWorkItem(Transaction tr, String lockId)

:base(new TryEnterLockDelegate(tr.TryEnterLock), lockId)

{

_Trans = tr;

_LockId = lockId;

}

internal WorkItem Copy()

{

return new CLockWorkItem(_Trans, _LockId);

}

}

}

using System;

using System.Collections.Generic;

using System.Linq;

using System.Text;

using System.IO;

namespace Cognitive.Docflow.Transactions

{

/// <summary>

/// Класс, описывающий файл, созданный в рамках транзакции

/// </summary>

public class TransactionFile : IDisposable

{

private String _id;

private FileStream _fs;

private Transaction _transaction;

private String _externalFullPath;

private String _internalFullPath;

private FileAccess _access;

/// <summary>

/// Копирует файл во временную директорию и работает с ним до завершения транзакции

/// </summary>

/// <param name="path">Полное имя файла (не пустое, если файл не существует, он будет создан после коммита)</param>

internal TransactionFile(string path)

{

if (String.IsNullOrEmpty(path))

throw new ArgumentException("Невозможно создать TransactionFile: полное имя файла не должно быть пустым.");

if (!Directory.Exists(Path.GetDirectoryName(path)))

throw new ArgumentException(String.Format("Невозможно создать TransactionFile {0}: его путь не существует!", path));

_transaction = Transaction.Current;

_internalFullPath = Path.GetTempFileName();

if (File.Exists(path))

File.Copy(path, _internalFullPath, true);

_externalFullPath = path;

_id = _externalFullPath + "||" + Guid.NewGuid().ToString();

_access = FileAccess.Read;

}

/// <summary>

/// Чистка

/// </summary>

~TransactionFile()

{

Dispose();

}

/// <summary>

/// Чистка

/// </summary>

public void Dispose()

{

if (_fs != null)

_fs.Dispose();

if (File.Exists(_internalFullPath))

File.Delete(_internalFullPath);

}

/// <summary>

/// идентификатор файла

/// </summary>

public String id {

get

{ return _id; }

}

/// <summary>

/// транзакция, к которой относится данный файл

/// </summary>

public Transaction transaction

{

get

{

return _transaction;

}

}

/// <summary>

/// возвращает поток для доступа к содержимому файла

/// </summary>

/// <param name="access"></param>

/// <returns></returns>

public FileStream GetStream(FileAccess access)

{

if (_fs == null || !(_fs.CanRead || _fs.CanWrite) || access != _access)

{

if (_fs != null)

_fs.Close();

_access = access;

if (!File.Exists(_internalFullPath) && File.Exists(_externalFullPath))

File.Copy(_externalFullPath, _internalFullPath);

_fs = File.Open(_internalFullPath, FileMode.OpenOrCreate, _access);

}

return _fs;

}

/// <summary>

/// закрывает открытый поток и сбрасываем содержимое на диск

/// (применяет к файлу все изменения)

/// </summary>

public void Flush()

{

///если производилась работа с потоком, то производим сброс изменений на диск

if (_fs != null)

_fs.Close();

if (!File.Exists(_internalFullPath))

return;

if (!Directory.Exists(Path.GetDirectoryName(_externalFullPath)))

throw new InvalidOperationException(String.Format("Невозможно записать содержимое TransactionFile на диск в файл {0}, т.к. пути его размещения не существует!!", _externalFullPath));

File.Copy(_internalFullPath, _externalFullPath, true);

File.Delete(_internalFullPath);

}

/// <summary>

/// возвращает полный путь к файлу

/// </summary>

public String FullPath { get { return _externalFullPath; } }

}

}

using System;

using System.Collections.Generic;

using System.Linq;

using System.Text;

using Cognitive.Docflow;

namespace Cognitive.Docflow.Environment

{

/// <summary>

/// серверная среда выполнения

/// </summary>

public abstract class EnvironmentServer

{

Transactions.ConnectionPool _cpool = null;

Cognitive.Threading.ThreadPool _tpool = null;

private BlockManager _blockManager = null;

private object _locker = new object();

private bool _stopping = false;

/// <summary>

/// конструктор серверной среды выполнения

/// </summary>

/// <param name="sqlConnectionString"></param>

protected EnvironmentServer(String sqlConnectionString)

{

//конструируем пул соединений

_cpool = new Transactions.ConnectionPool(sqlConnectionString, String.Empty, String.Empty);

//конструируем пул потоков

_tpool = new Cognitive.Threading.ThreadPool(4, TimeSpan.FromSeconds(1));

_blockManager = new BlockManager();

_tpool.OnIdle += new Cognitive.Threading.ThreadPoolEvent(ThreadPooll_OnIdle);

}

/// <summary>

/// конструктор серверной среды выполнения

/// </summary>

/// <param name="sqlConnectionString"></param>

/// <param name="threadsCount"></param>

protected EnvironmentServer(String sqlConnectionString, int threadsCount)

{

//конструируем пул соединений

_cpool = new Transactions.ConnectionPool(sqlConnectionString, String.Empty, String.Empty);

//конструируем пул потоков

_tpool = new Cognitive.Threading.ThreadPool(threadsCount, TimeSpan.FromMinutes(5));

_blockManager = new BlockManager();

_tpool.OnIdle += new Cognitive.Threading.ThreadPoolEvent(ThreadPooll_OnIdle);

}

internal BlockManager blockMan {

get { return _blockManager; }

}

private static Func<bool> _IdleProc

= new Func<bool>(() =>

{

if (Transactions.Transaction.Current.TryEnterLock("IDLE"))

return Transactions.Transaction.Current.Interaction.Idle();

return false;

});

private static Func<EnvironmentServer, Delegate, int> _TransactionCall =

new Func<EnvironmentServer, Delegate, int>((server, d) =>

{

Transactions.Transaction tr = server.OpenTransaction();

tr.Invoke(d);

tr.Commit();

return 0;

}

);

/// <summary>

/// запуск выполнения фоновых задач в рамках отдельной транзакции

/// </summary>

/// <param name="source"></param>

void ThreadPooll_OnIdle(Cognitive.Threading.ThreadPool source)

{

lock (_openTransactions)

{

if (_stopping)

return;

}

_TransactionCall.BeginInvoke(this, _IdleProc, null, null);

lock (_openTransactions)

{

foreach (Transactions.Transaction trans in _openTransactions)

{//Поиск транзакций слишком долго простаивающих и их откат, если они уже не откатываются и не накатываются

if (trans.MyIdleInterval > TimeSpan.FromMinutes(5)) //тайм-аут транзакции 5 минут

trans.Rollback();

}

}

}

/// <summary>

/// запуск рабочих потоков среды выполнения

/// </summary>

virtual public void Run()

{

}

/// <summary>

/// остановка рабочих потоков среды выполнения

/// </summary>

virtual public void Stop()

{

if (Threading.WorkingThread.Current != null)

throw new InvalidOperationException("Остановка сервера не может быть вызвана из рабочего потока");

lock (_openTransactions)

{

_stopping = true;

}

///проверяет есть ли еще открытые транзакции

Func<bool> HasOpenTrans = new Func<bool>(() =>

{

lock(_openTransactions)

{

return _openTransactions.Count > 0;

}

});

///возвращает последнюю открытую транзакцию

Func<Transactions.Transaction> CurrentTrans =

new Func<Transactions.Transaction>(() =>

{

lock (_openTransactions)

{

if (_openTransactions.Count > 0)

return _openTransactions[_openTransactions.Count - 1];

}

return null;

});

DateTime dtWaitFor = DateTime.Now + TimeSpan.FromMinutes(1);

while (HasOpenTrans() && (DateTime.Now < dtWaitFor))

System.Threading.Thread.Sleep(50);

///пока есть открытые транзакции - завершаем их

while (HasOpenTrans())

{

Transactions.Transaction curTrans = CurrentTrans();

if (curTrans == null)

continue;

curTrans.Rollback();

curTrans.CompleteHandle.WaitOne();

}

_tpool.Stop();

}

private List<Transactions.Transaction> _openTransactions =

new List<Cognitive.Docflow.Transactions.Transaction>();

/// <summary>

/// открытие транзакции

/// </summary>

/// <returns>созданный и проинициализированный объект транзакции</returns>

/// <remarks>

/// Создает транзакцию и связывает её с рабочим потоком.

/// Все открытае транзакции запоминаются.

/// В случае принудительной остановки сервера происходит откат всех незавершенных транзакций

/// </remarks>

virtual public Transactions.Transaction OpenTransaction()

{

Transactions.Transaction tr = Transactions.Transaction.CreateInstance(this, _cpool, _tpool);

lock (_openTransactions)

{

if (_stopping)

throw new InvalidOperationException("Нельзя создавать транзакции когда сервер находится в состоянии завершения");

_openTransactions.Add(tr);

}

tr.Complete += new Cognitive.Docflow.Transactions.TransactionEventHandler(tr_Complete);

return tr;

}

void tr_Complete(Cognitive.Docflow.Transactions.Transaction sender, EventArgs e)

{

lock (_openTransactions)

{

_openTransactions.Remove(sender);

}

}

public abstract IProtocolWriter GetProtocol(ActionPriority priority);

/// <summary>

/// конструирует корневой объект бизнес-логики

/// </summary>

/// <param name="datasource">соединение со средой хранения</param>

/// <returns></returns>

public abstract Environment.IDFContainerObject CreateRootObject(CTNKDESMLib.ICTDEDatabase datasource);

}

}

Размещено на Allbest.ru

...