egwmultienginemanager/EgwMultiEngineManager.Core/ExecProcessManager.vb

Imports System.Threading
Imports System.Windows.Threading
Imports EgwMultiEngineManager.Data
Imports Newtonsoft.Json

Public Class ExecProcessManager
   Implements IDisposable

   Public Event m_AllArgsProcessed()
   Public Event m_AnswerReceived(result As AnswerDTO)
   Public Event m_Statistics(Statistics As Statistics)

   Public Enum ExecutionThreadStatuses As Integer
      RUNNING = 1
      STOPPED = 2
   End Enum

   Private Enum ProcessManagerStates
      NULL = 0
      OPEN = 1
      CLOSE = 2
   End Enum

   Public Enum ReturnModes As Integer
      QUEUE = 1
      EVENT_ = 2
   End Enum

   Private Shared m_refEgtOutLog As Action(Of String)
   Public Shared Sub SetEgtOutLog(value As Action(Of String))
      m_refEgtOutLog = value
   End Sub

   Private m_StatisticsTimer As New DispatcherTimer(DispatcherPriority.Background)

   Private disposedValue As Boolean

   Private m_ProcessManagerState As ProcessManagerStates

   Private m_bReturnMode As ReturnModes

   Private m_nGroupId As Integer = 0

   Private m_ExecEnvironment As EXECENVIRONMENTS = EXECENVIRONMENTS.NULL
   Friend ReadOnly Property ExecEnvironment As EXECENVIRONMENTS
      Get
         Return m_ExecEnvironment
      End Get
   End Property

   ' riferimento al thread principale
   Private m_ExecutionThread As Thread
   Private m_ExecutionThreadStatus As ExecutionThreadStatuses = ExecutionThreadStatuses.STOPPED
   Public ReadOnly Property ExecutionThreadStatus As ExecutionThreadStatuses
      Get
         Return m_ExecutionThreadStatus
      End Get
   End Property
   ' array dei thread
   Private m_ThreadList As Thread()
   ' array dei dati dei thread
   Private m_ThreadDataList As ThreadData()
   Friend ReadOnly Property ThreadDataList As ThreadData()
      Get
         Return m_ThreadDataList
      End Get
   End Property
   ' variabile che ferma i processi
   Private m_bStopProcess As Boolean = False
   ' variabile che conferma chiusura dei processi
   Private m_bExecutionThreadStoped As Boolean = False
   ' numero massimo di istanze del cam
   Private m_MaxCamInstances As Integer = 1
   ' nome dell'eseguibile da lanciare
   Private m_sProcessFileName As String = ""
   ' stringa di argomenti dell'eseguibile da lanciare
   Private m_sProcessArguments As String = ""
   ' coda delle cose da eseguire
   Private ArgumentsQueueLock As New Object
   Private m_ArgumentsQueue As New Queue(Of QuestionDTO)
   Public ReadOnly Property ArgumentsQueue As Queue(Of QuestionDTO)
      Get
         Return m_ArgumentsQueue
      End Get
   End Property
   ' coda dei risultati
   Private ArgumentsResultQueueLock As New Object
   Private m_ArgumentsResultQueue As New Queue(Of AnswerDTO)
   Public ReadOnly Property ArgumentsResultQueue As Queue(Of AnswerDTO)
      Get
         Return m_ArgumentsResultQueue
      End Get
   End Property

   ' funzione che richiede prossimo elemento da processare
   Private m_delGetNextProcessArgs As Func(Of QuestionDTO)
   Public Sub SetGetNextProcessArgs(GetNextProcessArgs As Func(Of QuestionDTO))
      m_delGetNextProcessArgs = GetNextProcessArgs
   End Sub
   ' funzione di pre processing dell'elemento
   Private m_delPreProcess As Func(Of QuestionDTO, QuestionDTO)
   Public Sub SetPreProcess(PreProcess As Func(Of QuestionDTO, QuestionDTO))
      m_delPreProcess = PreProcess
   End Sub
   ' funzione di post processing dell'elemento
   Private m_delPostProcess As Action(Of AnswerDTO)
   Public Sub SetPostProcess(PostProcess As Action(Of AnswerDTO))
      m_delPostProcess = PostProcess
   End Sub

   Private m_nLastMinAnswerCounter As Integer = 0

   Public ReadOnly Property nQuestionInQueue As Integer
      Get
         Return ArgumentsQueueCount() + nCalculatingProcesses
      End Get
   End Property

   Public ReadOnly Property nRunningProcesses As Integer
      Get
         Return ThreadDataList.Count(Function(x) x IsNot Nothing)
      End Get
   End Property

   Public ReadOnly Property nCalculatingProcesses As Integer
      Get
         Return ThreadDataList.Count(Function(x) x IsNot Nothing AndAlso x.ProcessStatus = ThreadData.ProcessStatuses.WAITINGANSWER)
      End Get
   End Property

   Public ReadOnly Property nFreeProcesses As Integer
      Get
         Return ThreadDataList.Count(Function(x) x IsNot Nothing AndAlso x.ProcessStatus = ThreadData.ProcessStatuses.NULL)
      End Get
   End Property

   Public ReadOnly Property nFreeOrWaitingAnswerProcesses As Integer
      Get
         Return ThreadDataList.Count(Function(x) x IsNot Nothing AndAlso (x.ProcessStatus = ThreadData.ProcessStatuses.NULL OrElse x.ProcessStatus = ThreadData.ProcessStatuses.ANSWERRECEIVED))
      End Get
   End Property

   Private PerMinResultQueueLock As New Object
   Private m_PerMinResultQueue As New Queue(Of Integer)
   Public ReadOnly Property PerMinResultQueue As Queue(Of Integer)
      Get
         Return m_PerMinResultQueue
      End Get
   End Property

   Private m_bFirstStatistics As Boolean = False

   Private m_bDebug As Boolean = False
   Private m_bFirst As Boolean = True

#Region "CONSTRUCTORS"

   Sub New(nGroupId As Integer, nEnvironment As EXECENVIRONMENTS, sProcessFileName As String, sProcessArguments As String, nMaxCamInstances As Integer, bReturnMode As ReturnModes, bDebug As Boolean)
      m_nGroupId = nGroupId
      m_ExecEnvironment = nEnvironment
      m_sProcessFileName = sProcessFileName
      m_sProcessArguments = sProcessArguments
      SetMaxCamInstances(nMaxCamInstances)
      m_bReturnMode = bReturnMode
      m_bDebug = bDebug
      m_StatisticsTimer.Interval = TimeSpan.FromMinutes(1)
      AddHandler m_StatisticsTimer.Tick, AddressOf StatisticsTimer_Tick
      m_StatisticsTimer.Start()
   End Sub

#End Region    ' CONSTRUCTORS

   Friend Sub EgtOutLog(sMessage As String)
      If Not IsNothing(m_refEgtOutLog) Then
         m_refEgtOutLog(sMessage)
      End If
   End Sub

   Public Function ArgumentsEnqueue(ProcessArgs As QuestionDTO) As Boolean
      If ProcessArgs.nId <= 0 OrElse ProcessArgs.Args.Count = 0 Then Return False
      SyncLock ArgumentsQueueLock
         m_ArgumentsQueue.Enqueue(ProcessArgs)
      End SyncLock
      Return True
   End Function

   Public Function ArgumentsDequeue() As QuestionDTO
      Dim DequeueProcessArgs As QuestionDTO = Nothing
      SyncLock ArgumentsQueueLock
         If m_ArgumentsQueue.Count > 0 Then
            DequeueProcessArgs = m_ArgumentsQueue.Dequeue()
         End If
      End SyncLock
      Return DequeueProcessArgs
   End Function

   Public Function ArgumentsQueueCount() As Integer
      Dim nCount As Integer = 0
      SyncLock ArgumentsQueueLock
         nCount = m_ArgumentsQueue.Count
      End SyncLock
      Return nCount
   End Function

   Public Function ArgumentsResultEnqueue(ProcessArgsResult As AnswerDTO) As Boolean
      SyncLock ArgumentsResultQueueLock
         m_ArgumentsResultQueue.Enqueue(ProcessArgsResult)
      End SyncLock
      Return True
   End Function

   Public Function ArgumentsResultDequeue() As AnswerDTO
      Dim DequeueProcessArgsResult As AnswerDTO = Nothing
      SyncLock ArgumentsResultQueueLock
         If m_ArgumentsResultQueue.Count > 0 Then
            DequeueProcessArgsResult = m_ArgumentsResultQueue.Dequeue()
         End If
      End SyncLock
      Return DequeueProcessArgsResult
   End Function

   Public Function ArgumentsResultQueueCount() As Integer
      Dim nCount As Integer = 0
      SyncLock ArgumentsResultQueueLock
         nCount = m_ArgumentsResultQueue.Count
      End SyncLock
      Return nCount
   End Function

   Public Sub SetMaxCamInstances(nValue As Integer)
      ' Numero di core logici da utilizzare (minimo tra presenti sul PC e imposti da INI)
      Dim nMaxThread As Integer = Math.Min(Environment.ProcessorCount, nValue)
      m_MaxCamInstances = nMaxThread
   End Sub

   Private Sub StatisticsTimer_Tick()
      Dim nLastMinAnswer As Integer = 0
      Dim nLast5MinAnswer As Integer = 0
      Dim nLastHourAnswer As Integer = 0
      SyncLock PerMinResultQueueLock
         m_PerMinResultQueue.Enqueue(m_nLastMinAnswerCounter)
         If m_PerMinResultQueue.Count > 60 Then
            m_PerMinResultQueue.Dequeue()
         End If
         nLastMinAnswer = m_nLastMinAnswerCounter
         For nCount = m_PerMinResultQueue.Count To m_PerMinResultQueue.Count - 5 Step -1
            nLast5MinAnswer += m_PerMinResultQueue(nCount)
         Next
         nLastHourAnswer = m_PerMinResultQueue.Sum()
      End SyncLock
      m_nLastMinAnswerCounter = 0
      Dim CurrManagerStatistics As New Statistics(nRunningProcesses, nCalculatingProcesses, nLastMinAnswer, nLast5MinAnswer, nLastHourAnswer, nQuestionInQueue)
      RaiseEvent m_Statistics(CurrManagerStatistics)
   End Sub

   ' funzione che avvia thread principale
   Public Sub StartExecutionThread()
      If m_ExecutionThreadStatus = ExecutionThreadStatuses.RUNNING Then Return
      m_bStopProcess = False
      m_bExecutionThreadStoped = False
      m_ExecutionThread = New Thread(Sub()
                                        ExecutionProcess()
                                     End Sub)

      m_ExecutionThread.SetApartmentState(ApartmentState.STA)
      ' avvio thread di gestione della macchina che avvia la connessione
      m_ExecutionThread.Start()

      m_ExecutionThreadStatus = ExecutionThreadStatuses.RUNNING
      ' lancio apertura finestra statistiche processi
      m_ProcessManagerState = ProcessManagerStates.OPEN
   End Sub

   Public Sub StopExecutionThread()
      If m_ExecutionThreadStatus = ExecutionThreadStatuses.STOPPED Then Return
      m_bStopProcess = True
      While Not m_bExecutionThreadStoped
         Thread.Sleep(100)
      End While
      If m_ExecutionThread IsNot Nothing Then
         While Not (m_ExecutionThread.ThreadState = ThreadState.Aborted Or m_ExecutionThread.ThreadState = ThreadState.Stopped Or m_ExecutionThread.ThreadState = ThreadState.Suspended)
            Thread.Sleep(100)
         End While
         m_ExecutionThread = Nothing
      End If
      m_ThreadList = Nothing
      m_ExecutionThreadStatus = ExecutionThreadStatuses.STOPPED
      ' lancio chiusura finestra statistiche processi
      m_ProcessManagerState = ProcessManagerStates.CLOSE
   End Sub

   ' funzione di esecuzione del thread principale che gestice i processi
   Private Sub ExecutionProcess()
      Dim bStopMainProcess As Boolean = False
      Dim nStartingProc As Integer = 0
      While Not bStopMainProcess
         bStopMainProcess = m_bStopProcess
         ' se processo fermato, fermo i thread
         If bStopMainProcess Then
            If m_ThreadList IsNot Nothing AndAlso m_ThreadList.Count > 0 AndAlso m_ThreadList(0) IsNot Nothing Then
               ' li fermo
               m_bStopProcess = True
               ' verifico siano terminati
               Dim bOneNotEnded As Boolean = True
               While bOneNotEnded
                  bOneNotEnded = False
                  For Each Thread In m_ThreadList
                     If Thread IsNot Nothing Then
                        If Thread.IsAlive Then
                           bOneNotEnded = True
                        End If
                     End If
                  Next
               End While
               ' pulisco la lista
               For ThreadIndex = 0 To m_ThreadList.Count - 1
                  m_ThreadList(ThreadIndex) = Nothing
               Next
               m_bStopProcess = False
            End If
            m_bExecutionThreadStoped = True
            Return
         End If
         ' se lista processi nulla o vuota, li faccio partire
         If (m_ThreadList Is Nothing OrElse m_ThreadList.Count = 0) Then
            If Not m_bDebug OrElse m_bFirst Then
               m_bFirst = False
               m_ThreadList = New Thread(m_MaxCamInstances - 1) {}
               m_ThreadDataList = New ThreadData(m_MaxCamInstances - 1) {}
               For nThreadIndex = 0 To m_MaxCamInstances - 1
                  Dim ThreadId As Integer = nThreadIndex
                  m_ThreadList(nThreadIndex) = New Thread(Sub()
                                                             ThreadFunction(ThreadId)
                                                          End Sub)
                  m_ThreadList(nThreadIndex).SetApartmentState(ApartmentState.STA)
                  ' avvio thread di gestione della macchina che avvia la connessione
                  m_ThreadList(nThreadIndex).Start()
                  Thread.Sleep(100)
               Next
            End If
         End If
         ' se qualche processo in stop, lo faccio ripartire
         If Not m_bDebug Then
            For ThreadIndex = 0 To m_ThreadList.Count - 1
               If ThreadIndex < m_ThreadList.Count Then
                  Dim Thread = m_ThreadList(ThreadIndex)
                  If Thread IsNot Nothing Then
                     If Thread.ThreadState = ThreadState.Stopped OrElse
                     Thread.ThreadState = ThreadState.Aborted OrElse
                     Thread.ThreadState = ThreadState.Suspended OrElse
                     (Not (m_ThreadDataList Is Nothing) AndAlso Not m_ThreadDataList(ThreadIndex).ProcessStatus = ThreadData.ProcessStatuses.TOBESTARTED AndAlso
                     ((m_ThreadDataList(ThreadIndex).Process Is Nothing) OrElse
                     ((m_ThreadDataList(ThreadIndex).Process IsNot Nothing) AndAlso m_ThreadDataList(ThreadIndex).Process.HasExited))) Then
                        ' inserire un ritardo di rilancio?
                        ' lo chiudo e rilancio
                        'Thread.Sleep(500)
                        'Thread.Abort()
                        If Not (m_ThreadDataList(ThreadIndex).Process Is Nothing) Then
                           Try
                              m_ThreadDataList(ThreadIndex).Process.Kill()
                           Catch ex As Exception
                           End Try
                        End If
                        Thread.Sleep(100)
                        While Not (Thread.ThreadState = ThreadState.Aborted Or Thread.ThreadState = ThreadState.Stopped Or Thread.ThreadState = ThreadState.Suspended)
                           Thread.Sleep(100)
                        End While
                        Thread = Nothing
                        Dim ThreadId As Integer = ThreadIndex
                        m_ThreadList(ThreadIndex) = New Thread(Sub()
                                                                  ThreadFunction(ThreadId)
                                                               End Sub)
                        m_ThreadList(ThreadIndex).SetApartmentState(ApartmentState.STA)
                        ' avvio thread di gestione della macchina che avvia la connessione
                        m_ThreadList(ThreadIndex).Start()
                     End If
                  Else
                     Dim ThreadId As Integer = ThreadIndex
                     If ThreadIndex < m_ThreadList.Count Then
                        m_ThreadList(ThreadIndex) = New Thread(Sub()
                                                                  ThreadFunction(ThreadId)
                                                               End Sub)
                        m_ThreadList(ThreadIndex).SetApartmentState(ApartmentState.STA)
                        ' avvio thread di gestione della macchina che avvia la connessione
                        m_ThreadList(ThreadIndex).Start()
                     End If
                  End If
               End If
            Next
         End If
         ' verifico se coda vuota e tutti i processi hanno finito
         'Dim nQueueArgs As Integer = 0
         'SyncLock ArgumentsQueueLock
         '   nQueueArgs = m_ArgumentsQueue.Count
         'End SyncLock
         'If nQueueArgs = 0 Then
         '   Dim bProcessWaiting As Boolean = False
         '   For Each ThreadData In m_ThreadDataList
         '      Dim CurrThreadData = ThreadData
         '      If Not IsNothing(CurrThreadData) OrElse CurrThreadData.ProcessStatus = ThreadData.ProcessStatuses.WAITINGANSWER OrElse CurrThreadData.ProcessStatus = ThreadData.ProcessStatuses.ANSWERRECEIVED Then
         '         bProcessWaiting = True
         '         Exit For
         '      End If
         '   Next
         '   If Not bProcessWaiting Then
         '      ' creo thread per lanciare evento asincrono di fine calcolo
         '      Dim AllArgsProcessedThread = New Thread(Sub()
         '                                                 RaiseEvent m_AllArgsProcessed()
         '                                              End Sub)
         '      AllArgsProcessedThread.SetApartmentState(ApartmentState.STA)
         '      AllArgsProcessedThread.Start()
         '   End If
         'End If
         If Not m_bFirstStatistics Then
            m_bFirstStatistics = True
            RaiseEvent m_Statistics(New Statistics(nRunningProcesses, nCalculatingProcesses, 0, 0, 0, 0))
         End If
         Thread.Sleep(1000)
      End While
   End Sub

   ' funzione di esecuzione del singolo processo che comunica con il programma
   Private Sub ThreadFunction(ThreadIndex As Integer)
      Dim MyThreadData As New ThreadData
      m_ThreadDataList(ThreadIndex) = MyThreadData

      ' avvio processo
      Dim ProcInfo As New ProcessStartInfo(m_sProcessFileName, m_nGroupId.ToString() & ThreadIndex.ToString() & " " & m_sProcessArguments)
      ProcInfo.RedirectStandardInput = True
      ProcInfo.RedirectStandardOutput = True
      ProcInfo.RedirectStandardError = True
      ProcInfo.UseShellExecute = False
      ProcInfo.CreateNoWindow = True

      Dim Proc = Process.Start(ProcInfo)
      AddHandler Proc.OutputDataReceived, AddressOf Thread_OutputDataReceived
      AddHandler Proc.ErrorDataReceived, AddressOf Thread_ErrorDataReceived
      If Not (Proc Is Nothing) Then
         Proc.BeginOutputReadLine()
         Proc.BeginErrorReadLine()
         MyThreadData.SetProcess(Proc)
         MyThreadData.SetProcessStatus(ThreadData.ProcessStatuses.NULL)
         ' ciclo per leggere coda ed eseguire
         While Not m_bStopProcess AndAlso Not Proc.HasExited
            Select Case MyThreadData.ProcessStatus
               Case ThreadData.ProcessStatuses.NULL
                  MyThreadData.SetProcessStatus(ThreadData.ProcessStatuses.TAKINGREQUEST)
                  Dim NextProcessArgs As QuestionDTO = Nothing
                  If m_delGetNextProcessArgs IsNot Nothing Then
                     NextProcessArgs = m_delGetNextProcessArgs()
                  Else
                     NextProcessArgs = ArgumentsDequeue()
                  End If
                  If NextProcessArgs IsNot Nothing AndAlso NextProcessArgs.nId > 0 AndAlso Not NextProcessArgs.Args.Count = 0 Then
                     If m_delPreProcess IsNot Nothing Then
                        NextProcessArgs = m_delPreProcess(NextProcessArgs)
                     End If
                     MyThreadData.SetCurrRequest(NextProcessArgs)
                     NextProcessArgs.SetThreadIndex(ThreadIndex)
                     Proc.StandardInput.WriteLine("#8477271#" & NextProcessArgs.sProcessArgs & "#8477271#")
                     Proc.StandardInput.Flush()
                     MyThreadData.SetProcessStatus(ThreadData.ProcessStatuses.WAITINGANSWER)
                  Else
                     MyThreadData.SetProcessStatus(ThreadData.ProcessStatuses.NULL)
                  End If
               Case ThreadData.ProcessStatuses.WAITINGANSWER
                  Thread.Sleep(10)
               Case ThreadData.ProcessStatuses.ANSWERRECEIVED
                  If m_delPostProcess IsNot Nothing Then
                     m_delPostProcess(MyThreadData.nProcResult)
                  End If
                  m_nLastMinAnswerCounter += 1
                  Select Case m_bReturnMode
                     Case ReturnModes.QUEUE
                        ArgumentsResultEnqueue(MyThreadData.nProcResult)
                     Case ReturnModes.EVENT_
                        RaiseEvent m_AnswerReceived(MyThreadData.nProcResult)
                  End Select
                  MyThreadData.SetProcessStatus(ThreadData.ProcessStatuses.NULL)
            End Select
            Threading.Thread.Sleep(10)
         End While
         If m_bStopProcess Then
            Proc.StandardInput.WriteLine("quit")
            Proc.StandardInput.Flush()
            RemoveHandler Proc.OutputDataReceived, AddressOf Thread_OutputDataReceived
            RemoveHandler Proc.ErrorDataReceived, AddressOf Thread_ErrorDataReceived
         End If
      End If
      MyThreadData.SetProcess(Nothing)
   End Sub

   Private Sub Thread_OutputDataReceived(sender As Object, e As DataReceivedEventArgs)
      Dim sResult As String = e.Data
      If Not String.IsNullOrWhiteSpace(sResult) AndAlso sResult.StartsWith("#8376261#") AndAlso sResult.EndsWith("#8376261#") Then
         sResult = sResult.Substring(9, sResult.Length - 18)
         Dim Result As AnswerDTO = JsonConvert.DeserializeObject(Of AnswerDTO)(sResult)
         m_ThreadDataList(Result.nThreadIndex).SetProcResult(Result)
         m_ThreadDataList(Result.nThreadIndex).SetProcessStatus(ThreadData.ProcessStatuses.ANSWERRECEIVED)
      Else
         EgtOutLog("Riga su StandardOutput non riconosciuta " & e.Data & " GroupId=" & m_nGroupId)
      End If
   End Sub

   Private Sub Thread_ErrorDataReceived(sender As Object, e As DataReceivedEventArgs)
      EgtOutLog("Errore su StandardError! " & e.Data & " GroupId=" & m_nGroupId)
   End Sub

   Protected Overridable Sub Dispose(disposing As Boolean)
      If Not disposedValue Then
         If disposing Then
            ' TODO: dispose managed state (managed objects)
         End If

         ' TODO: free unmanaged resources (unmanaged objects) and override finalizer
         ' TODO: set large fields to null
         StopExecutionThread()
         disposedValue = True
      End If
   End Sub

   ' ' TODO: override finalizer only if 'Dispose(disposing As Boolean)' has code to free unmanaged resources
   ' Protected Overrides Sub Finalize()
   '     ' Do not change this code. Put cleanup code in 'Dispose(disposing As Boolean)' method
   '     Dispose(disposing:=False)
   '     MyBase.Finalize()
   ' End Sub

   Public Sub Dispose() Implements IDisposable.Dispose
      ' Do not change this code. Put cleanup code in 'Dispose(disposing As Boolean)' method
      Dispose(disposing:=True)
      GC.SuppressFinalize(Me)
   End Sub

End Class