CSharp:Multithread Tips vari

Per fare debug vedere suggerimenti: Tips dall'IDE di Visual Studio

Avvio

Esempi di avvio

protected volatile bool IsRunning = false;
 
protected override bool StartDataCollectors()
{
	IsRunning = true;
 
	ParameterizedThreadStart th = new ParameterizedThreadStart(MainThread);
	Thread mainTh = new Thread(th);
	mainTh.Start(); // Nonostante abbia usato 'ParameterizedThreadStart' NON passo alcun parametro 
 
	return true;
}
 
private void MainThread(object data)
{
	Logger.Info(string.Format("STARTING thread for analyze complete xml file, ship={0}-{1}", avShip.CompanyCode, avShip.ShipCode, communicationMode));
	ParameterizedThreadStart th = new ParameterizedThreadStart(ManageShip);
	lstThreadShip[counter] = new Thread(th);
	lstThreadShip[counter].Start(new object[] { avShip.CompanyCode, avShip.ShipCode, counter, communicationMode });
 
	// ...
}
 
private void ManageShip(object param)
{
	while (IsRunning)
	{
		// ....
	}
}

Avvio passando un parametro

_redisKeepAlive = new Thread(new ParameterizedThreadStart(KeepAlive));
_redisKeepAlive.Start(new object[] { _redisKeepAlive.ManagedThreadId }); // <---- !!!
 
// ... fine del metodo
 
private void KeepAlive(object param)
{
	while (true)
	{
		try
		{			
			object[] paramArray = (object[])param;			// <---- !!!
			int threadId = (int)paramArray[0];		// <---- !!!
 
			bool locked = _wfHelper.LockExtend(_redisKey, _redisKey, _redisLockExpirySec);
 
			if (!IsRunning)
				break; 
 
			if (!locked)
			{
				Logger.Error("KeepAlive() - Can't extend Redis cache value! Restarting...");
				Stop();
				Thread.Sleep(_redisLockExpirySec * 1000);
				Start();
				break;
			}
			else
			{
				_dicSubThreadsIsRunning.Add("", true);
			}
		}
		catch (Exception exc)
		{
			Logger.ErrorException(String.Format("KeepAlive() - Error extending lock: "), exc);
		}
 
		Thread.Sleep(_redisLockExpirySec * 1000 / 3);
	}
}

Parallel Invoke

Esecuzione di Task paralleli senza indicazione di priorità:

using System.Threading.Tasks;
...
 
Task task2 = Task.Factory.StartNew(() =>
{
 
	Parallel.Invoke(
		//new ParallelOptions() { MaxDegreeOfParallelism = 4 },
		() => LoadLocationCodesScheduler(),
		() => LoadMediaTypeList(),
		() => LoadCampaignList(),
		() => LoadAudiologistList(),
		() => LoadInternList(),
		() => LoadGenderList(),
		() => LoadServiceAppList(),
		() => LoadRoomList(),
		() => LoadSpecialSlots(null),
		() => LoadRestrictionAndAppointement(),
		() => LoadRestriction(),
		() => GetParentServiceList(CurrentShopModel.SHOP_CODE)
		);
});
 
Task.WaitAll(task2);

Task paralleli con priorità

Esempio qui

Gestione risorse

Controllo del tempo

Per impostare un ritardo di esecuzione (in millisecondi):

using System.Threading;
 
//... etc...
Thread.Sleep(_redisLockExpirySec * 1000 ); // Ritardo in MILLISECONDI

Un Task del Thread per volta

Per avere un ciclo di lavorazione che va ripetuto ogni N secondi. E vogliamo evitare parallelismi, per cui se un ciclo di lavorazione dura più di N non voglio che ne parta uno nuovo in parallelo ma voglio attendere finché la lavorazione precedente non sia finita.

// Fields di instanza, fuori dal loop
volatile bool isEnabled;   // qualcuno prima di lanciare il thread lo setta a true
int loopIntervalSecs;             // numero di secondi del ciclo, lo prendo da qualche parte
volatile bool isRunning;
//
 
 
// Qui inizia il codice del mio thread
isRunning = true;
int waitTime = loopIntervalSecs * 1000;
 
DateTime nextTick = DateTime.UtcNow;
 
while (isEnabled)
{
	if (DateTime.UtcNow > nextTick)
	{
		// Qui faccio il mio lavoro...
		DoTheJob();
 
		nextTick = nextTick.AddMilliseconds(waitTime);
 
		if (nextTick < DateTime.UtcNow) nextTick = DateTime.UtcNow;
	}
 
	Thread.Sleep(500);
}
 
isRunning = false;

Controlli

Elenco threads

Per avere l'elenco di TUTTI i threads:

using System.Diagnostics;
 
ProcessThreadCollection currentThreads = Process.GetCurrentProcess().Threads;
 
foreach (ProcessThread thread in currentThreads)    
{
   // Verifiche del caso sul singolo thread
}

Esecuzione con Timeout

Da stackoverflow:

using System.Threading;
 
class Program {
    static void DoSomething() {
        try {
            // your call here...
            obj.PerformInitTransaction();         
        } catch (ThreadAbortException) {
            // cleanup code, if needed...
        }
    }
 
    public static void Main(params string[] args) {
 
        Thread t = new Thread(DoSomething);
        t.Start();
        if (!t.Join(TimeSpan.FromSeconds(30))) {
            t.Abort();
            throw new Exception("More than 30 secs.");
        }
    }
}

Invece se serve solo conoscere se si è travalicato il limite temporale

static void Main(string[] args)
{
	try
	{
		//etc
 
		if (m_MaxSecExecTime > 0) //Se c'è un TimeOut!
		{
			m_logger.Debug("Execution with timeout {0}s", m_MaxSecExecTime);
			var taskSendEmailSMS = Task.Run(() =>
			{
				DoWork_4SP(PMSMECtrl, logInfo, customerPhoneNr
						 , m_XMLEmailBackupInfoPath, m_DigitalkNodeCode);
			});
			bool isWithinTimeMax = taskSendEmailSMS.Wait(TimeSpan.FromSeconds(m_MaxSecExecTime));
			if (isWithinTimeMax)
			{
				m_logger.Debug("End. Task performed within '{0}' seconds ({1}s)."
								, m_MaxSecExecTime
								, DateTime.Now.Subtract(dtStart).TotalSeconds);
			}
			else
			{
				string strError = string.Format("\r\nSending e-mail+SMS takes longer than max time allowed. ({0}s)"
												, DateTime.Now.Subtract(dtStart).TotalSeconds);
				//m_logger.Error(strError);
				throw new TimeoutException(strError);
			}
		}
		else
		{
			DoWork_4SP(PMSMECtrl, logInfo, customerPhoneNr
					 , m_XMLEmailBackupInfoPath, m_DigitalkNodeCode);
			m_logger.Debug("End Task without timeout");
		}
	}
	catch (Exception ex)
	{
		m_logger.Error("Exception for ActionGateway 'StoredProcedure', fatal error:\r\n{0}", ex.Message);
		throw;
	}
}
 
 
private static void DoWork_4SP(PMSMEController PMSMECtrl, T_D_SMSEMAIL_Log logInfo
							 , string customerPhoneNr
							 , string XMLEmailBackupInfoPath, string digitalkNodeCode)
{
	try
	{
		//TEST comportamento in caso di eccezione:
		//      throw new Exception("Test2 comportamento in caso di eccezione");
		PMSMECtrl.Send_EmailSMS(logInfo, customerPhoneNr
								, XMLEmailBackupInfoPath
								, digitalkNodeCode);
	}
	catch (Exception ex)
	{
		m_logger.Error("Exception for ActionGateway 'StoredProcedure', fatal error:\r\n{0}", ex.Message);
		throw;
	}
}

Variabili

Volatile

Vedere qui: variabili x multithreading
DOC volatile

le variabili per il controllo d'esecuzione dei subThread conviene dichiararle 'volatile'

protected volatile bool IsRunning = false;
 
        protected override bool StopDataCollectors()
        {
            IsRunning = false;
 
            try
            {
// Manca controllo che i thread abbiano effettivamente finito il loro lavoro
                if (_wfHelper != null)
                { 
                    _wfHelper.LockRelease(_redisKey, _redisKey);
                }
            }
            catch (Exception ex)
            {
                Logger.ErrorException("Exception stop collector", ex);
            }            
            return true;
        }

Worker

Interazione con la UI

BackgroundWorker worker = new BackgroundWorker();
worker.DoWork += delegate(object s, DoWorkEventArgs args)
{
	try
	{
		// some stuff
 
		// Valorizzazione di variabili all'esterno del worker !!!!
		this.Dispatcher.Invoke(DispatcherPriority.Background,
			new Action(() =>
			{
				//btnInProgressFilterProductCode.IsEnabled = IS_DDT_SEARCH_ENABLEB;
				btnInProgressFilterDDTCode.IsEnabled = IS_DDT_SEARCH_ENABLEB;
			}));
		// some stuff
	}
	catch (Exception ex)
	{
		this.Dispatcher.Invoke(new Action(() =>
				RadMessageBox.Show(ex.message, LocalizationCommonProvider.GetString("EXCEPTION"),
					MessageBoxButton.OK, MessageBoxImage.Error)
		));
	}
}
 
// varie
worker.RunWorkerCompleted += delegate(object s, RunWorkerCompletedEventArgs args)
{
	// Al termine del worker, azioni specifiche
}
 
worker.RunWorkerAsync(); // (----------- lancia il worker

Varie

ID del Thread

All'interno di un Thread è possibile ottenerne l'ID numerico:

int threadId = Thread.CurrentThread.ManagedThreadId;

Oppure all'atto della dichiarazione del Thread e successivo start, ecco il Thread ID:

ParameterizedThreadStart satParamTh = new ParameterizedThreadStart(ManageSatellite);
Thread thSat = new Thread(satParamTh);
thSat.Start(new object[] { thSat.ManagedThreadId });   // <-- Thread ID !!!!!!!!!!!!!!!!!

Nome del thread

E' possibile all'interno del metodo adibito a thread settare la variabile Name:

Thread.CurrentThread.Name = "ThreadUDPListner";

Uso del Timer

Link interno: Usare il timer che all'interno della pagina Creare un servizio Windows

Chiudere (Kill) tutti i Thread

Ci son vari modi di chiudere\uccidere\killare tutti i Thread.

Usare la proprietà IsBackground

Questa consentirà al sistema operativo (.Net framework) di occuparsi della chiusura, questo è molto consigliato.

Usare metodo Abort

/// <summary>
/// TUTTI i Thread effettivamente lanciati nel MainThread per ciascuno c'è il flag IsRunning.
/// </summary>
private static List<Thread> _lstAllThreds = new List<Thread>();
 
 
private void MainThread(object data)
{
	//...
	// Es di popolamento lista threads
	_redisKeepAlive = new Thread(new ThreadStart(KeepAlive));
	_lstAllThreds.Add(_redisKeepAlive);
	_redisKeepAlive.Start();
	//..
}
 
protected override bool StopAllThreadsDataCollector()
{
	IsRunning = false;
	DateTime stopFromTime = DateTime.Now;
	int millisecondToWaitIfAlive = 700;
 
	try
	{
		Logger.Info("Checking the stop all of '{0}' Threads.", _lstAllThreds.Count);
		// Verifica che TUTTI i subThreads abbiano concluso il proprio lavoro compiutamente
		// ToDo inserire un limite temporale ed ucciderli tutti!!!!!
		bool allStopped = false;
		while (!allStopped)
		{
			allStopped = true;
			foreach (Thread t in _lstAllThreds)
			{
				if (t.IsAlive)
				{
					//Thread.CurrentThread.Name
					Logger.Warn("Thread [{0}] still running! Sleep {1} milliseconds...", 
								t.ManagedThreadId, millisecondToWaitIfAlive);
					allStopped = false;
					Thread.Sleep(millisecondToWaitIfAlive); // Attende mezzo secondo
				}
				else
				{
					Logger.Debug("Thread [{0}] correctly closed.", t.ManagedThreadId);
				}
			}
 
			if (Math.Abs(DateTime.Now.Subtract(stopFromTime).TotalMinutes) > MaxMinutesToStopThreads)
			{
				// Per killare i processi altra strada poteva essere dichiararli IsBackground
				Logger.Error("Timeout stopping all pending Thread (TimeOut = {0} minutes)\r\nKILLING ALL THREAD!", MaxMinutesToStopThreads);
				foreach (Thread t in _lstAllThreds)
				{
					if (t.IsAlive)
					{
						t.Abort();
					}
				}
				allStopped = true;
			}
		} // while (!allStopped)
 
		// Si toglie il Lock
		if (_wfHelper != null)
		{ 
			_wfHelper.LockRelease(_redisKey, _redisKey);
		}
		Logger.Info("All Thread stopped End.");
	}
	catch (Exception ex)
	{
		Logger.ErrorException("Exception stopping XmlCollector", ex);
	}            
	return true;
}

Usare la Join

Se associato al metodo che si occupa della chiusura questo NON terminerà sinchè tutti i sui thread non saranno chiusi.Occhio alla documentazione Microsoft
Blocca il thread chiamante finché non termina un thread o finché non trascorre l'intervallo di tempo specificato, pur continuando a eseguire la distribuzione SendMessage e COM standard.

protected override bool StopAllThreadsDataCollector()
{
	IsRunning = false;
 
	//sveglio i thread in ascolto sulle porte tcp per farli uscire
	allDone_Backup.Set();
	allDone_Complete.Set();
	allDone_Partial.Set();
 
	if (lstTcpListnerThread != null)
	{
		foreach (Thread th in lstTcpListnerThread)
		{
			th.Join();
		}
	}
	foreach (Thread th in lstThreadShip)
	{
		th.Join();
	}
	foreach (Thread th in lstThreadShipPartialData)
	{
		th.Join();
	}
	foreach (Thread th in lstThreadRadar)
	{
		if (th != null)
			th.Join();
	}
	if (satelliteThread != null)
	{
		satelliteThread.Join();
	}
 
	_redisManager.LockRelease(_redisKey, _machineName);
 
	return true;
}

Si può usare la Join e definire un timeout e se questo scatta si può eseguire l'Abort, segue esempio da stackoverflow:

using System.Threading;
 
class Program {
    static void DoSomething() {
        try {
            // your call here...
            obj.PerformInitTransaction();         
        } catch (ThreadAbortException) {
            // cleanup code, if needed...
        }
    }
 
    public static void Main(params string[] args) {
 
        Thread t = new Thread(DoSomething);
        t.Start();
        if (!t.Join(TimeSpan.FromSeconds(30))) {
            t.Abort();
            throw new Exception("More than 30 secs.");
        }
    }
}

Altro esempio preso da StackOverflow:

List<Thread> workerThreads = new List<Thread>();
List<int> results = new List<int>();
 
for (int i = 0; i < 5; i++) {
    Thread thread = new Thread(() => {
        Thread.Sleep(new Random().Next(1000, 5000));
        lock (results) {
            results.Add(new Random().Next(1, 10));
        }
    });
    workerThreads.Add(thread);
    thread.Start();
}
 
// Wait for all the threads to finish so that the results list is populated.
// If a thread is already finished when Join is called, Join will return immediately.
foreach (Thread thread in workerThreads) {
    thread.Join();
}

Debug.WriteLine("Sum of results: " + results.Sum());

Mappa e Links

---

C# | Multithreading

---

Visual Studio