Threadpool/WaitHandle 리소스 누출/충돌

https://stackoverflow.com/questions/3268606

16-09-2020
|

문제

내 디자인을 다시 생각해볼 필요가 있을 것 같아요.컴퓨터가 완전히 중단되고 때로는 VS 2010에서 HRESULT 0x8007000E가 발생하는 버그를 찾는 데 어려움을 겪고 있습니다.

데이터베이스 대기열을 기반으로 파일 전송을 처리하는 콘솔 응용 프로그램(나중에 서비스로 변환할 예정)이 있습니다.

전송이 허용된 스레드를 제한하고 있습니다.이는 우리가 연결하는 일부 시스템이 특정 계정의 특정 수의 연결만 포함할 수 있기 때문입니다.

예를 들어, 시스템 A는 3개의 동시 연결(3개의 개별 스레드를 의미함)만 허용할 수 있습니다.이러한 스레드 각각에는 고유한 연결 개체가 있으므로 연결을 공유하지 않으므로 동기화 문제가 발생해서는 안 됩니다.

우리는 해당 시스템의 파일을 주기적으로 처리하고 싶습니다.예를 들어 연결당 최대 100개의 파일을 전송할 수 있는 3개의 연결을 허용합니다.즉, 시스템 A에서 1000개의 파일을 이동하려면 3개의 스레드에 각각 100개의 파일이 허용되므로 주기당 300개의 파일만 처리할 수 있습니다.따라서 이 전송 기간 동안 우리는 10개의 스레드를 갖게 됩니다.한 번에 3개만 실행할 수 있습니다.따라서 3개의 주기가 있으며 마지막 주기에서는 마지막 100개의 파일을 전송하는 데 1개의 스레드만 사용합니다.(3개 스레드 x 100개 파일 = 주기당 300개 파일)

예를 들어 현재 아키텍처는 다음과 같습니다.

System.Threading.Timer는 GetScheduledTask()를 호출하여 5초마다 대기열에서 수행할 작업이 있는지 확인합니다.
아무것도 없으면 GetScheduledTask()는 아무 작업도 수행하지 않습니다.
작업이 있으면 ThreadPool 스레드를 생성하여 해당 작업을 처리 [작업 스레드 A]
작업 스레드 A는 전송할 파일이 1000개임을 확인합니다.
작업 스레드 A는 파일을 가져오는 시스템에서 실행 중인 스레드가 3개만 있을 수 있음을 확인합니다.
작업 스레드 A는 세 개의 새로운 작업 스레드 [B,C,D]를 시작하고 전송합니다.
작업 스레드 A는 B,C,D를 기다립니다. [WaitHandle.WaitAll(transfersArray)]
작업 스레드 A는 대기열에 아직 더 많은 파일이 있음을 확인합니다(현재는 700이어야 함).
작업 스레드 A는 대기할 새 배열을 생성합니다. [transfersArray = new TransferArray[3] 이는 시스템 A의 최대값이지만 시스템에 따라 다를 수 있습니다.
작업 스레드 A는 세 개의 새로운 작업 스레드 [B,C,D]를 시작하고 이를 기다립니다. [WaitHandle.WaitAll(transfersArray)]
더 이상 이동할 파일이 없을 때까지 프로세스가 반복됩니다.
작업 스레드 A는 완료되었음을 나타냅니다.

신호를 처리하기 위해 ManualResetEvent를 사용하고 있습니다.

내 질문은 다음과 같습니다

내가 겪고 있는 리소스 누출이나 문제를 일으킬 수 있는 눈에 띄는 상황이 있습니까?
매번 배열을 반복해야 합니까? WaitHandle.WaitAll(array) 그리고 전화해 array[index].Dispose()?
이 프로세스에 대한 작업 관리자 아래의 핸들 수가 천천히 늘어납니다.
System.Threading.Timer에서 작업자 스레드 A의 초기 생성을 호출합니다.이것에 문제가 있을까요?해당 타이머의 코드는 다음과 같습니다.

(일정 예약을 위한 일부 수업 코드)

private ManualResetEvent _ResetEvent;

private void Start()
{
    _IsAlive = true;
    ManualResetEvent transferResetEvent = new ManualResetEvent(false);
    //Set the scheduler timer to 5 second intervals
    _ScheduledTasks = new Timer(new TimerCallback(ScheduledTasks_Tick), transferResetEvent, 200, 5000);
}

private void ScheduledTasks_Tick(object state)
{
    ManualResetEvent resetEvent = null;
    try
    {
        resetEvent = (ManualResetEvent)state;
        //Block timer until GetScheduledTasks() finishes
        _ScheduledTasks.Change(Timeout.Infinite, Timeout.Infinite);
        GetScheduledTasks();
    }
    finally
    {
        _ScheduledTasks.Change(5000, 5000);
        Console.WriteLine("{0} [Main] GetScheduledTasks() finished", DateTime.Now.ToString("MMddyy HH:mm:ss:fff"));
        resetEvent.Set();
    }
}


private void GetScheduledTask()
{
    try 
    { 
        //Check to see if the database connection is still up
        if (!_IsAlive)
        {
            //Handle
            _ConnectionLostNotification = true;
            return;
        }

        //Get scheduled records from the database
        ISchedulerTask task = null;

        using (DataTable dt = FastSql.ExecuteDataTable(
                _ConnectionString, "hidden for security", System.Data.CommandType.StoredProcedure,
                new List<FastSqlParam>() { new FastSqlParam(ParameterDirection.Input, SqlDbType.VarChar, "@ProcessMachineName", Environment.MachineName) })) //call to static class
        {
            if (dt != null)
            {
                if (dt.Rows.Count == 1)
                {  //Only 1 row is allowed
                    DataRow dr = dt.Rows[0];

                    //Get task information
                    TransferParam.TaskType taskType = (TransferParam.TaskType)Enum.Parse(typeof(TransferParam.TaskType), dr["TaskTypeId"].ToString());
                    task = ScheduledTaskFactory.CreateScheduledTask(taskType);

                    task.Description = dr["Description"].ToString();
                    task.IsEnabled = (bool)dr["IsEnabled"];
                    task.IsProcessing = (bool)dr["IsProcessing"];
                    task.IsManualLaunch = (bool)dr["IsManualLaunch"];
                    task.ProcessMachineName = dr["ProcessMachineName"].ToString();
                    task.NextRun = (DateTime)dr["NextRun"];
                    task.PostProcessNotification = (bool)dr["NotifyPostProcess"];
                    task.PreProcessNotification = (bool)dr["NotifyPreProcess"];
                    task.Priority = (TransferParam.Priority)Enum.Parse(typeof(TransferParam.SystemType), dr["PriorityId"].ToString());
                    task.SleepMinutes = (int)dr["SleepMinutes"];
                    task.ScheduleId = (int)dr["ScheduleId"];
                    task.CurrentRuns = (int)dr["CurrentRuns"];
                    task.TotalRuns = (int)dr["TotalRuns"];

                    SchedulerTask scheduledTask = new SchedulerTask(new ManualResetEvent(false), task);
                    //Queue up task to worker thread and start
                    ThreadPool.QueueUserWorkItem(new WaitCallback(this.ThreadProc), scheduledTask);     
                }
            }
        }

    }
    catch (Exception ex)
    {
        //Handle
    }
}

private void ThreadProc(object taskObject)
{
    SchedulerTask task = (SchedulerTask)taskObject;
    ScheduledTaskEngine engine = null;
    try
    {
        engine = SchedulerTaskEngineFactory.CreateTaskEngine(task.Task, _ConnectionString);
        engine.StartTask(task.Task);    
    }
    catch (Exception ex)
    {
        //Handle
    }
    finally
    {
        task.TaskResetEvent.Set();
        task.TaskResetEvent.Dispose();
    }
}

해결책 4

이 이상한 문제의 원인은 아키텍처와 관련된 것이 아니라 솔루션을 3.5에서 4.0으로 변환했기 때문인 것으로 나타났습니다.코드를 변경하지 않고 솔루션을 다시 만들었으며 문제가 다시는 발생하지 않았습니다.

다른 팁

0x8007000E는 메모리 부족 오류입니다.그것과 핸들 수는 리소스 누출을 가리키는 것 같습니다.구현하는 모든 객체를 폐기하고 있는지 확인하세요. IDisposable.여기에는 다음 배열이 포함됩니다. ManualResetEvent당신이 사용하고 있습니다.

시간이 있으면 .NET 4.0을 사용하여 변환할 수도 있습니다. Task 수업;이와 같은 복잡한 시나리오를 훨씬 더 깔끔하게 처리하도록 설계되었습니다.자식을 정의하여 Task 개체를 사용하면 전체 스레드 수를 줄일 수 있습니다(스레드는 예약뿐만 아니라 스택 공간 때문에 비용이 많이 듭니다).

비슷한 문제(시간이 지남에 따라 핸들 수가 증가함)에 대한 답변을 찾고 있습니다.

귀하의 애플리케이션 아키텍처를 살펴보고 도움이 될 수 있는 사항을 제안하고 싶습니다.

IOCP(입력 출력 완료 포트)에 대해 들어보셨나요?

C#을 사용하여 이를 구현하는 것이 어려운지 잘 모르겠지만 C/C++에서는 매우 쉽습니다.이를 사용하면 고유 스레드 풀 (해당 풀의 스레드 수는 일반적으로 2 x로 정의되어 있습니다. PC 또는 서버의 프로세서 또는 프로세서 수)이 풀을 IOCP 핸들에 연결하면 풀이 수행합니다. 일하다.다음 기능에 대한 도움말을 참조하세요.CreateIoCompletionPort();PostQueuedCompletionStatus();GetQueuedCompletionStatus();

일반적으로 스레드를 즉시 생성하고 종료하는 것은 시간이 많이 걸릴 수 있으며 성능 저하 및 메모리 조각화로 이어질 수 있습니다.MSDN과 Google에는 IOCP에 관한 수천 개의 문헌이 있습니다.

나는 당신의 아키텍처를 완전히 재고해야 한다고 생각합니다.동시에 3개의 연결만 가질 수 있다는 사실은 파일 목록을 생성하는 데 1개의 스레드를 사용하고 파일을 처리하는 데 3개의 스레드를 사용해야 한다는 것을 거의 요구합니다.생산자 스레드는 모든 파일을 대기열에 삽입하고 3개의 소비자 스레드는 대기열에서 항목이 도착하면 대기열에서 제외되고 계속 처리됩니다.차단 대기열은 코드를 크게 단순화할 수 있습니다..NET 4.0을 사용하는 경우 다음을 활용할 수 있습니다. BlockingCollection 수업.

public class Example
{
    private BlockingCollection<string> m_Queue = new BlockingCollection<string>();

    public void Start()
    {
        var threads = new Thread[] 
            { 
                new Thread(Producer), 
                new Thread(Consumer), 
                new Thread(Consumer), 
                new Thread(Consumer) 
            };
        foreach (Thread thread in threads)
        {
            thread.Start();
        }
    }

    private void Producer()
    {
        while (true)
        {
            Thread.Sleep(TimeSpan.FromSeconds(5));
            ScheduledTask task = GetScheduledTask();
            if (task != null)
            {
                foreach (string file in task.Files)
                {
                    m_Queue.Add(task);
                }
            }
        }
    }

    private void Consumer()
    {
        // Make a connection to the resource that is assigned to this thread only.
        while (true)
        {
            string file = m_Queue.Take();
            // Process the file.
        }
    }
}

위의 예에서는 확실히 내용을 지나치게 단순화했지만 일반적인 아이디어를 얻으셨기를 바랍니다.스레드 동기화 방식이 많지 않고(대부분 차단 대기열에 포함됨) 물론 다음을 사용하지 않기 때문에 이것이 훨씬 더 간단하다는 점에 유의하십시오. WaitHandle 사물.분명히 스레드를 정상적으로 종료하려면 올바른 메커니즘을 추가해야 하지만 이는 매우 쉽습니다.

라이센스 : CC-BY-SA ~와 함께 속성

제휴하지 않습니다 StackOverflow