Why the number of threads is continuously increasing with SerialPort.CatchReceivedEvents()

Question

I´m running a C# application with .NET Framework 2.0 to read data from SerialPort to get the weight from a scale.

The application works fine, does what it is supposed to do, but the number of threads keeps increasing and more memory is consumed until the application crashes, usually after around 4 hours.

When running with a serialport simulator the number of threads is stable around 30. But when I use an actual scale it goes greater than 500 threads.

I used Microsoft Managed Stack Explorer 1.0 to take a dump of the threads and almost all of them have exactly the following stack:

0. System.IO.Ports.SerialPort.CatchReceivedEvents (Source Unavailable)
1. System.IO.Ports.SerialStream.EventLoopRunner.CallReceiveEvents (Source Unavailable)
2. System.Threading._ThreadPoolWaitCallback.WaitCallback_Context (Source Unavailable)
3. System.Threading.ExecutionContext.Run (Source Unavailable)
4. System.Threading._ThreadPoolWaitCallback.PerformWaitCallbackInternal (Source Unavailable)
5. System.Threading._ThreadPoolWaitCallback.PerformWaitCallback (Source Unavailable)

I'm not able to identify the reason these threads are being created. Does anybody have any idea of what I'm missing here? Thanks!

This is my code:
Scale.cs -> creates a thread when method open() is called. The thread reads the value from getWeight().
Scales.cs -> treats events from serial port in method SerialPort_DataReceived(...). It's where m_SerialPort.ReadLine() is called and ends up providing the value to getWeight().

Scale.cs:

    using System;
    using System.Collections.Generic;
    using System.Threading;
    using ScalesGSS;
    using StateMachine.Exceptions;
    using StateMachine.Log;
    using StateMachine.MessageOutput;

    namespace StateMachine.DriverImplementation
    {

    class Scale : AScale
    {
        private const int Scale_version = 1;

        private Thread thread = null;

        private IScales gScale = null;

        //
        private string m_Type;
        private string m_PortName;
        private int m_BaudRate;
        private char m_Parity;
        private int m_DataBits;
        private string m_StopBits;
        private int m_CommandReturnLength;
        private string m_CommandType;
        private string m_CommandValue;
        private int m_ReadTimeOutInMilliseconds;
        private int m_WeightInitialPosition;
        private int m_WeightFinalPosition;
        private int m_TimeBetweenReadsInMilliseconds;
        private int m_StableReadQuantity;
        private int m_MinimumWeight;
        private int m_ScaleID;
        //
        private double m_OldWeight = 0.0;
        private double m_Offset = 0.0;
        private double m_CurrentWeight = 0.0;
        int m_WeightEqualCount = 0;
        //
        byte m_Status = 3; // "NO COMMUNICATION"
        //
        private bool m_Closed = false;
        private static LogFactory m_Log = new LogFactory(LogCategory.Device, "");
        ErrorDialog m_ErrorDialog = new ErrorDialog();

        public Scale()
        {
            this.setClassName("Scale");
            this.setDeviceType(DeviceType.Scale);
        }

        public void run()
        {
            try
            {

                if (this.m_Type.ToUpper().Equals("GENERICSCALES")) // GENERICSCALES or MOCKSCALES
                    this.gScale = new ScalesGSS.GenericScales();
                else
                    this.gScale = new ScalesGSS.MockScales();

                this.gScale.PortName = this.m_PortName;
                this.gScale.BaudRate = this.m_BaudRate;
                this.gScale.Parity = this.m_Parity.ToString();
                this.gScale.DataBits = this.m_DataBits;
                this.gScale.StopBits = this.m_StopBits;
                this.gScale.CommandReturnLength = this.m_CommandReturnLength;
                this.gScale.CommandType = this.m_CommandType;
                this.gScale.CommandValue = this.m_CommandValue;
                this.gScale.ReadTimeOut = this.m_ReadTimeOutInMilliseconds;
                this.gScale.WeightInitialPosition = this.m_WeightInitialPosition;
                this.gScale.WeightFinalPosition = this.m_WeightFinalPosition;
                this.gScale.setParameters();
                this.gScale.configurePort();

                while (true)
                {
                    if (this.m_Closed)
                    {
                        if (this.OpenedPort())
                            this.gScale.closePort();
                        break;
                    }

                    Thread.Sleep(this.m_TimeBetweenReadsInMilliseconds);

                    if (!this.OpenedPort())
                    {
                        if (!this.OpenPort())
                        {
                            m_Log.writeLogWarning("Error opening serialport.", " Port: " + this.m_PortName, true);
                        }
                    }

                    if (this.ErrorReadingWeight())
                    {
                        m_Log.writeLogWarning("Invalid weight.", " Port: " + this.m_PortName, true);
                    }

                    this.m_CurrentWeight = getWeight();

                    if (!ReadingTimeout())
                    {
                        if (this.m_WeightEqualCount > m_StableReadQuantity)
                        {
                            if (m_CurrentWeight > m_MinimumWeight)
                                m_Status = 2; // "WEIGHT STABLE"
                            else
                            {
                                m_Status = 0; // "SCALE FREE"
                                m_WeightEqualCount = 0;
                            }
                        }
                        else
                        {
                            if (m_CurrentWeight > m_MinimumWeight)
                            {
                                m_Status = 1; // "STABILIZING"

                                if ((this.m_CurrentWeight >= (this.m_OldWeight - this.m_Offset)) && (this.m_CurrentWeight <= (this.m_OldWeight + this.m_Offset)))
                                    this.m_WeightEqualCount++;
                                else
                                    this.m_WeightEqualCount = 0;

                                this.m_OldWeight = this.m_CurrentWeight;
                            }
                            else
                            {
                                m_Status = 0; // "SCALE FREE"
                                m_WeightEqualCount = 0;
                            }
                        }
                    }
                    else
                    {
                        m_WeightEqualCount = 0;
                        m_Status = 3;         // "NO COMMUNICATION"
                        string v_Message = "No communication with scale. Port: " + m_PortName;
                        m_Log.writeLogWarning(v_Message, "", true);
                        AutoClosingMessageBox.Show(v_Message, "Scale", 10000);
                    }
                }
            }
            catch (Exception v_Exception)
            {
                m_Log.writeLogError("run()", v_Exception);
            }
        }

        private bool OpenedPort()
        {
            return this.gScale.OpenedPort;
        }

        private bool OpenPort()
        {
            bool v_OpenPort;
            v_OpenPort = this.gScale.openPort();

            if (!v_OpenPort)
            {
                m_ErrorDialog.getScaleErrorMessage(gScale);
            }

            return v_OpenPort;
        }

        private bool ErrorReadingWeight()
        {
            return this.gScale.ErrorReadingWeight;
        }

        private double getWeight()
        {
            return this.gScale.getWeight();
        }

        private DateTime LastGoodReading()
        {
            return gScale.LastGoodReading;
        }

        private void setLastGoodReading(DateTime p_Value)
        {
            gScale.LastGoodReading = p_Value;
        }

        private bool ReadingTimeout()
        {
            if (m_ReadTimeOutInMilliseconds > 0)
            {
                DateTime v_LastGoodReading = LastGoodReading() == DateTime.MinValue ? DateTime.Now : LastGoodReading();
                setLastGoodReading(DateTime.Now);
                return DateTime.Now > v_LastGoodReading.AddMilliseconds(m_ReadTimeOutInMilliseconds);
            }
            else
                return false;
        }

        #region "IDriverService"

        public override byte getStatus()
        {
            return m_Status;
        }

        public override byte[] read()
        {
            return System.Text.ASCIIEncoding.ASCII.GetBytes(m_CurrentWeight.ToString());
        }

        public override byte[] read(int p_InitialPosition, int p_Size)
        {
            return read();
        }

        public override byte[] write(byte[] p_Data)
        {
            string v_Temp = System.Text.ASCIIEncoding.ASCII.GetString(p_Data);

            if (v_Temp.Equals("getScaleNumber"))
                return System.Text.ASCIIEncoding.ASCII.GetBytes(m_ScaleID.ToString());
            else
                throw new EDriverAccess(1, "Not implemented");
        }

        public override bool open()
        {
            this.thread = new Thread(run);
            this.thread.Name = "SCALE";
            this.thread.IsBackground = true;
            this.thread.Start();
            return true;
        }

        public override bool close()
        {
            try
            {
                this.release();
                return true;
            }
            catch
            {
                return false;
            }
        }

        public override int getVersion()
        {
            return Scale_version;
        }

        public override void setProperties(Dictionary<string, string> p_props)
        {
            try
            {
                this.m_Type = p_props["type"];
                this.m_PortName = p_props["portName"];
                this.m_BaudRate = Int32.Parse(p_props["baudRate"]);
                this.m_Parity = char.Parse(p_props["parity"]);
                this.m_DataBits = Int32.Parse(p_props["dataBits"]);
                this.m_StopBits = p_props["stopBits"];
                this.m_CommandReturnLength = Int32.Parse(p_props["returnLength"]);
                this.m_CommandType = p_props["commandType"];
                this.m_CommandValue = p_props["commandValue"];
                this.m_ReadTimeOutInMilliseconds = Int32.Parse(p_props["readTimeout"]);
                this.m_WeightInitialPosition = Int32.Parse(p_props["weightInitPos"]);
                this.m_WeightFinalPosition = Int32.Parse(p_props["weightFinPos"]);
                this.m_TimeBetweenReadsInMilliseconds = Int32.Parse(p_props["delayLeitura"]);
                this.m_StableReadQuantity = Int32.Parse(p_props["qtdeLeituraEstavel"]);
                this.m_MinimumWeight = Int32.Parse(p_props["pesoMinimo"]);
                this.m_ScaleID = Int32.Parse(p_props["numBalanca"]);
                if (p_props.ContainsKey("precision"))
                    this.m_Offset = Int32.Parse(p_props["precision"]);
            }
            catch (Exception)
            {
                throw new Exception();
            }
        }

        public override void release()
        {
            this.m_Closed = true;
            m_Status = 3; // "NO COMMUNICATION"
        }
        #endregion
    }
}

Scales.cs:

using System;
using System.Collections.Generic;
using System.IO;
using System.IO.Ports;
using System.Reflection;
using System.Timers;
using Scales.Util;


namespace Scales.DLL
{
    public class Scales : Status
    {
        public event EventHandler StableWeightChanged;

        protected virtual void OnCountdownCompleted(EventArgs e)
        {
            if (StableWeightChanged != null)
                StableWeightChanged(this, e);

        }

        System.Timers.Timer timerTimeWithoutSample;
        private int m_IntervalsWithoutSample = 0;
        private string m_EndOfWeightChar = "";

        private void _timer_Elapsed(object sender, ElapsedEventArgs e)
        {
            m_IntervalsWithoutSample++;
        }

        public int IntervalsWithoutSample { get { return m_IntervalsWithoutSample; } }

        private SerialPort m_SerialPort;

        public string PortName { get; set; }
        public int BaudRate { get; set; }
        public int DataBits { get; set; }

        private Double m_Weight = 0;
        public Double Weight
        {
            get
            {
                if (m_BufferWeights.Count > 0)
                {
                    try
                    {
                        m_Weight = treatReceivedValue(m_BufferWeights[m_BufferWeights.Count - 1]);
                    }
                    catch
                    {

                    }
                    finally
                    {
                        ErrorReadingWeight = (m_Weight != -1 ? false : true);
                    }

                }
                else
                {
                    m_Weight = 0;
                }
                return m_Weight;
            }
        }

        public List<Double> getAndFlushPastWeights()
        {
            List<Double> v_FlushedValues = new List<double>();

            Double v_WeightCursor;
            while (m_BufferWeights.Count > 1 && v_FlushedValues.Count < 200)
            {
                v_WeightCursor = treatReceivedValue(m_BufferWeights[0]);

                if (v_WeightCursor >= 0)
                {
                    v_FlushedValues.Add(v_WeightCursor);
                }

                m_BufferWeights.RemoveAt(0);
            }
            return v_FlushedValues;
        }

        public void ResetWeights()
        {
            if (m_BufferWeights != null)
            {
                m_BufferWeights.Clear();
            }
        }

        public string NewLineCommandType { get; set; }
        public string NewLineCommand { get; set; }
        public int ReturnLength { get; set; }
        public int WeightInitialPosition { get; set; }
        public int WeightFinalPosition { get; set; }
        public int MotionBitPos { get; set; }

        public int ReadTimeOut { get; set; }
        public bool OpenedPort { get; private set; }
        public bool ErrorReadingWeight { get; private set; }
        public DateTime LastGoodReading { get; private set; }

        public bool IsStable { get; private set; }

        private Parity PortParity { get; set; }
        public string SerialParity
        {
            get { return PortParity.ToString(); }
            set
            {
                setParity(value);
            }
        }

        public int WeightReadLength
        {
            get
            {
                if (WeightFinalPosition >= WeightInitialPosition)
                {
                    return WeightFinalPosition - WeightInitialPosition + 1;
                }
                else
                {
                    return 0;
                }
            }
        }

        private StopBits PortStopBits { get; set; }
        public string SerialStopBits
        {
            get { return PortStopBits.ToString(); }
            set
            {
                setStopBits(value);
            }
        }

        private void setParity(string p_Parity)
        {
            if (p_Parity.Equals(Parity.Even.ToString()))
            {
                PortParity = Parity.Even;
            }
            else if (p_Parity.Equals(Parity.Mark.ToString()))
            {
                PortParity = Parity.Mark;
            }
            else if (p_Parity.Equals(Parity.Odd.ToString()))
            {
                PortParity = Parity.Odd;
            }
            else if (p_Parity.Equals(Parity.Space.ToString()))
            {
                PortParity = Parity.Space;
            }
            else
            {
                PortParity = Parity.None;
            }
        }

        private void setStopBits(string p_StopBits)
        {
            if (p_StopBits.Equals(StopBits.One.ToString()))
            {
                PortStopBits = StopBits.One;
            }
            else if (p_StopBits.Equals(StopBits.OnePointFive.ToString()))
            {
                PortStopBits = StopBits.OnePointFive;
            }
            else if (p_StopBits.Equals(StopBits.Two.ToString()))
            {
                PortStopBits = StopBits.Two;
            }
            else if (p_StopBits.Equals("1"))
            {
                PortStopBits = StopBits.One;
            }
            else if (p_StopBits.Equals("1.5"))
            {
                PortStopBits = StopBits.OnePointFive;
            }
            else if (p_StopBits.Equals("2"))
            {
                PortStopBits = StopBits.Two;
            }
            else
            {
                PortStopBits = StopBits.None;
            }
        }

        public Scales()
        {
            OpenedPort = false;
            ErrorReadingWeight = false;
            IsStable = false;
            m_IntervalsWithoutSample = 999999;
            timerTimeWithoutSample = new System.Timers.Timer(5);
            timerTimeWithoutSample.Elapsed += new ElapsedEventHandler(_timer_Elapsed);
        }

        private int ignoreNextXValues;
        public void resetScale()
        {
            ErrorReadingWeight = false;
            IsStable = false;
            m_IntervalsWithoutSample = 999999;
            ignoreNextXValues = 2;

            m_BufferWeights.Clear();
            m_BufferTime.Clear();

            if (m_SerialPort != null && m_SerialPort.IsOpen)
            {
                m_SerialPort.Close();
                m_SerialPort.Open();
                m_SerialPort.DiscardInBuffer();
            }

        }

        List<String> m_BufferWeights = new List<String>();
        List<String> m_BufferTime = new List<String>();

        public bool openPort()
        {
            try
            {
                if (m_SerialPort.IsOpen)
                {
                    m_SerialPort.Close();
                }

                m_SerialPort.Open();
                resetScale();

                OpenedPort = true;
                return true;
            }
            catch (Exception ex)
            {
                MessageDetail = ex.Message;
                Return = -100;
                OpenedPort = false;
                return false;
            }
        }

        public bool closePort()
        {
            try
            {
                if (m_SerialPort != null)
                {
                    if (m_SerialPort.IsOpen)
                    {
                        m_SerialPort.Close();
                    }
                }
                OpenedPort = false;

                return true;
            }
            catch (Exception ex)
            {
                MessageDetail = ex.Message;
                Return = -101;
                return false;
            }
        }

        public bool configurePort()
        {
            try
            {
                m_SerialPort = new SerialPort();
                m_SerialPort.PortName = PortName;
                m_SerialPort.BaudRate = BaudRate;
                m_SerialPort.Parity = PortParity;
                m_SerialPort.DataBits = DataBits;
                m_SerialPort.StopBits = PortStopBits;
                m_SerialPort.ReadTimeout = ReadTimeOut > 0 ? ReadTimeOut : SerialPort.InfiniteTimeout;
                m_SerialPort.NewLine = getNewLineCommand();
                m_SerialPort.DataReceived += new SerialDataReceivedEventHandler(SerialPort_DataReceived);

                return true;
            }
            catch (Exception ex)
            {
                MessageDetail = ex.Message;
                Return = -102;
                return false;
            }
        }

        private string getNewLineCommand()
        {
            string v_Command = string.Empty;

            if (NewLineCommandType.ToUpper().Equals(CommandTypes.CHAR.ToUpper()))
            {
                byte v_Char = Convert.ToByte(NewLineCommand);
                v_Command = Convert.ToChar(v_Char).ToString();
            }
            else if (NewLineCommandType.ToUpper().Equals(CommandTypes.STRING.ToUpper()))
            {
                v_Command = NewLineCommand;
            }
            else
            {
                char[] v_delimiters = { '|' };
                String[] v_Strings = NewLineCommand.Split(v_delimiters);

                if (v_Strings.Length == 2)
                {
                    v_Command = v_Strings[0];
                    m_EndOfWeightChar = v_Strings[1];
                }
                else
                {
                    v_Command = NewLineCommand;
                }
            }

            return v_Command;
        }

        private void SerialPort_DataReceived(object sender, SerialDataReceivedEventArgs e)
        {
            try
            {
                LastGoodReading = DateTime.Now;
                string ReadLine = m_SerialPort.ReadLine();
                m_BufferWeights.Add(ReadLine);
            }
            catch (Exception)
            {
                m_Weight = 0;
                LastGoodReading = DateTime.MinValue;
            }
        }

        private Double treatReceivedValue(string p_ReceivedValue)
        {
            try
            {
                if (ignoreNextXValues > 0) ignoreNextXValues--;
                if (ignoreNextXValues > 0) return 0;

                double v_Value = double.MinValue;
                p_ReceivedValue = p_ReceivedValue.Replace("\r", "").Replace("\n", "");

                m_IntervalsWithoutSample = 0;

                if (p_ReceivedValue.Length < WeightInitialPosition + WeightReadLength)
                {
                    return -1;
                }
                if (MotionBitPos != -1 && p_ReceivedValue.Length < MotionBitPos - 1)
                {
                    return -1;
                }

                string strValor = "";

                if (NewLineCommandType.ToUpper().Equals(CommandTypes.VARIABLE_LENGTH.ToUpper()))
                {
                    int v_EndCharPos = p_ReceivedValue.IndexOf(m_EndOfWeightChar);

                    if (v_EndCharPos != -1)
                    {
                        strValor = p_ReceivedValue.Substring(0, v_EndCharPos).Trim();
                    }
                }
                else
                {
                    strValor = p_ReceivedValue.Substring(WeightInitialPosition, WeightReadLength).Trim();
                }

                bool IsDouble = double.TryParse(strValor, out v_Value);

                if (IsDouble)
                {
                    if (MotionBitPos != -1)
                    {
                        string bit = p_ReceivedValue.Substring(MotionBitPos, 1).Trim();
                        if (bit == "1")
                        {
                            IsStable = true;
                        }
                        else IsStable = false;
                    }
                    else
                    {
                        IsStable = true;
                    }

                    return v_Value;
                }
                else
                {
                    return -1;
                }
            }
            catch (Exception ex)
            {
                Return = -200;
                MessageDetail = ex.Message + " - Fonte:readScales";
                ErrorReadingWeight = true;
            }
            return -1;
        }
    }
}

Answer 1

You should try to reduce your problematic code down to something more manageable, as it will make it easier for others to debug. There's a lot of application logic in there that's probably not relevant to the problem which can make it hard for people to see what's going on. You'll get a lot more answers if your example is shorter. You may even figure the problem out yourself in the process!

Having said that, I have a hunch about what's wrong but you'll need to do a little bit of the leg-work yourself to discover if I'm right or wrong:

The .NET serial port works by waiting for data to come in, and then firing the DataReceived event on a worker thread whenever it notices that there's new data. I believe you have 400 or 500 of these worker threads that never complete their work, so they never go away.

Your event handler for the SerialPort.DataReceived event looks like it's blocking waiting for a whole line to come in, but the event is going to be fired whenever there's some amount of new data on the serial port (not necessarily a whole line). If a long line of text comes in, the DataReceived event is going to fire many times, each on it's own worker thread. These worker threads are synchronized to each other, so they're all going to wait for the previous one to finish.

1. The first thread that gets queued up is going to wait for a while at m_SerialPort.ReadLine() until the whole line comes in.
2. A bunch of threads queue up behind the first thread as more characters come in. The rest of the threads will end up waiting for the first thread to finish running your event handler.
3. Finally, the whole line comes in. The first thread finishes, and one of the 5 or 6 that are queued up behind it gets to run and the process starts all over.
4. The running thread blocks on ReadLine, 5 or 6 more queue up behind it. (We're now back at 1)

Eventually you have so many threads queued up that you run into memory issues.

You probably have the read-timeout on m_SerialPort set to timeout.Infinite. If you set the timeout to something smaller, like 1 second (1000) and you get a lot of TimeoutExceptions in your SerialPort_DataReceived method, then I'm probably right

Side Note You should catch a more specific exception type in your DataReceived event handler. Catching exception can mask exactly this type of problem.

If I've correctly diagnosed the problem, you'll need to change the architecture of your program a little bit. The simplest thing to do is not subscribe to the DataReceived event and to have a single worker thread call m_SerialPort.ReadLine(); with an infinite timeout. When it reads a line, have that worker thread raise an event with the whole line of text received and subscribe to THAT event instead of the SerialPort.DataReceived(); event.

Alternatively, if you want to subscribe to the SerialPort.DataReceived(); event, then read individual characters out of the SerialPort until SerialPort.BytesToRead is zero and stick them in a buffer. Then, when you have a whole line raise some "LineReceived" event that you make yourself that returns the whole line at once as one of the EventArgs. This method is doesn't require you to spool up your own thread that persists for a really long time.

Answer 2

I had a similar problem, using SerialPort.ReadExisting() insted of SerialPort.ReadLine() I was able to avoid the creation of infinite threads.