Thread pool mantığı nasıl kurulmalı ?

[mrmarman]

Merhaba.

* Forumlardaki başlıklara bakarken aklıma geldi. Bir proje tasarlayarak merakımı gidermeye karar verdim. Thread içerikli ve eksiklerimi pekiştirmek amaçlı.
* En altta düşündüğüm ve çalışacağına inancımın tam olduğu bir çözüm de var ama konuyu özetlemiş olayım isterim :

* Örnek sorunsal çekirdeği vereyim : 
* 13 tane dosya indireceğiz. ( Özellikle 13 dedim bir tane 10'lu bir tane 3'lü thread beklemesi ile küsuratlı olsun istedim... )
* Ağ sağlığı için en fazla 10 thread aynı anda çalışsın istiyoruz. ( Bu bile fazla ama örnekleme kolay olsun diye böyle diyelim. Mantığı kurunca ekonomisi yapılır...)
* Yalnız bir sonraki sekans için 10 thread'in 10'unu da bitene kadar beklemeyelim, 10'dan bitenin yerine kalan 3'den bir tanesi hemen devreye girsin. Çünkü 10 thread içerisinde en yavaş olan thread hızı tüm bekleme süresini belirleyecektir bunu biliyoruz.
 

* Genel uygulamam şu şekildedir. 
• Bir ana proje MainThread çalışıyor.
  
• Bu ana projeye işleri üstelenecek bir Worker thread eklerim. Böylece ana proje dahili işlemlerde bloke olmuyor. Eventleri Syncronize vb. ile çalıştırıp haberleştiriyoruz vs.. 
  
• Ardışık olarak internetten bir sürü dosya indirmek istiyoruz. 
  
• Bunun için WaitForMultipleObject'i başarılı bir şekilde kullanıyoruz. Mesela sadece 10 tane dosya simültane indirsin, tümü bitince sonraki 10 tane dosya indirme işlemine başlasın şekilde yürüyoruz.
  

Aklıma gelen soru bu noktada... 

• Thread sayısını sınırlandırdığımda (10 tane Threadhandle'ini bir Array ile vererek) WaitForMultipleObject ile (TRUE ile hepsi bitince talimatıyla)  tüm dosyalar indiğinde WAIT_OBJECT_0 şeklinde bir sonuç ile işlem başarılı bir şekilde sonuçlanarak bir sonraki sekansa geçmek için sebebim oluyor.
  
• Eğer hepsi bitince talimatını FALSE yaparsam önce hangisi biterse onun index nosu ile ( WAIT_OBJECT_0+1, WAIT_OBJECT_0+5, vb. ) array içerisinden hangisinin işi önce bitmişse onun index nosunu alıyorum.
  

İşte sorum bu noktada...   

10 Thread'in 10'unu da beklemeden bitenin yerine yenisini başlatmak istersem hemen altında yeni bir WaitFor kurmak için nasıl bir strateji kurmak gerekli.

- Düşüncemde yer eden çözüm; Recursive bir yapı ile "waitformultipleobject" / waitfor all = "FALSE" olarak  ilkinde 10 diğerlerinde tek elemanlı alt alta dallanan bir single çalışan multiobject yapısı gibi bir beklenti.

- Recursive derken kendi içinde kendini çağıran ama ilk çağrıda maksimum thread sayısı iken içe dallandığında daima 1 elemanlı array ile bekleyen bir yapı.

* Bu noktada çözmem gereken sorunsal ise alta dallandıktan sonra yine bir WaitFor koyup devam etmekte olan, ARRAY içindeki tüm thread'lerin bittiğinden de emin olmak için nasıl bir yapı kuracağım...?! Sanki Array içinde kalan elemanları da SUB WaitFor için gönderip onların da waitfor'a dahil olmasını beklemek gibi bir garip düşünce de oluşmadı değil...

* Bu şekilde çözebileceğimi biliyorum ve inanıyorum ama kolları sıvamadan sorayım dedim.... Mantıklı mıdır yoksa alternarifi var mıdır ? 

* Başlığa yabancılık çekenler için anahtar kelimeleri vereyim :
* WaitForSingleObject 1 tread sonlanana kadar procedure bekletir ve bitince procedure kaldığı yerden devam eder.
* WaitForMultipleObjects  ( n ) adet thread bekler...  bWaitAll= TRUE ise tümü bitmediği sürece bekler procedure de bekler, bWaitAll = FALSE ise en az birisi biterse, bekleme işlemi sona erer ama diğer threadler çalışmaya devam ederken procedure yeniden yürümeye başlar.

[vkamadan]

Muharrem bey merhaba ,
Tarif ettiğiniz iş için en doğru senkronizasyon yapısısının SEMAPHORE olacağını düşünüyorum, özet yapı olarak,  Semaphore u bir thread odası gibi düşünün içeride aynı anda çalıştırılacak thread sayısını belirliyorsunuz örn 10 , biten iş parçacığı yerine bekleyen yenisini  alıp işletiyor. Özel bir tecrübem olmadığı için kod paylaşımıyorum ama anahtar kelime vermiş olayım.
iyi çalışmalar.

Düzeltme ;
Size bu mesajı gönderdikten sonra Semaphore ile ilgili @Tuğrul HELVACI beyin  bana hazırladığı örneğe denk geldim , hemen paylaşıyorum. Örnekte 4 adet Thread var Semaphore 2 boyutuna sahip olarak oluşturulduğu için  , 4 thread aynı anda Start ediliyor ancak  beklendiği gibi aynı anda 2 si çalışacak şekilde senkronize oluyorlar.

unit uTest;

interface

uses
 Winapi.Windows, Winapi.Messages, System.SysUtils, System.Variants, System.Classes, Vcl.Graphics,
 Vcl.Controls, Vcl.Forms, Vcl.Dialogs,

 SyncObjs,
 CodeSiteLogging, Vcl.StdCtrls;

type
 TForm1 = class(TForm)
   Button1: TButton;
   procedure Button1Click(Sender: TObject);
 private
   ASemaphore : TSemaphore;

 public
   { Public declarations }
 end;


var
 Form1: TForm1;
 A,B,C,D,E : TThread;


implementation
uses
 System.Diagnostics;

{$R *.dfm}





procedure TForm1.Button1Click(Sender: TObject);
begin
 ASemaphore := TSemaphore.Create(nil, 2, 2, '');

 A :=
 TThread.CreateAnonymousThread(
   procedure
   begin
     TThread.CurrentThread.FreeOnTerminate := false;

     CodeSite.Send('Thread A Created and running');

     WaitForSingleObject(ASemaphore.Handle, INFINITE);
     Sleep(5000);
     ASemaphore.Release(1);

     CodeSite.Send('Thread A Exiting');
   end
 );

 B :=
 TThread.CreateAnonymousThread(
   procedure
   begin
     CodeSite.Send('Thread B Created and running');

     WaitForSingleObject(ASemaphore.Handle, INFINITE);
     Sleep(5000);
     ASemaphore.Release(1);

     CodeSite.Send('Thread B Exiting');
   end
 );

 C:=
 TThread.CreateAnonymousThread(
   procedure
   begin
     CodeSite.Send('Thread C Created and running');

     WaitForSingleObject(ASemaphore.Handle, INFINITE);
     Sleep(5000);
     ASemaphore.Release(1);

     CodeSite.Send('Thread C Exiting');
   end
 );

 D:=
 TThread.CreateAnonymousThread(
   procedure
   begin
     CodeSite.Send('Thread D Created and running');

     WaitForSingleObject(ASemaphore.Handle, INFINITE);
     Sleep(5000);
     ASemaphore.Release(1);

     CodeSite.Send('Thread D Exiting');
   end
 );

 A.Start;
 B.Start;
 C.Start;
 D.Start;


end;

end.

[Tuğrul HELVACI]

Yoğun bir tempoda çalıştığım için bu aralar, pek dikkatlice okuyamadım ama; anladığım kadarı ile çözüm getirmeye çalışayım. Öncelikle; Delphi'nin yeni versiyonlarında thread pool sınıfları var. Ama siz kendiniz implemente etmek isterseniz; belirli bir sayıda thread'i (best practice olarak, çekirdek sayısı * 2) başlatarak bir thread güvenlikli queue ile (Örneğin TThreadQueue) çalışabilirsiniz.

Kuyrukta bir şey olmadığı durumda, çalışan tüm threadler kuyruktan bir item çekemeyecek dolayısı ile bekler durumda olacaklardır. Kuyruğa bir şey geldiğinde kuyruğu consume etmeye çalışan 10 thread'den şanslı olanı kuyruktan item'i çekecek ve kuyruk yeniden boş kalacağı için 9 adet thread'de kuyruğa bir şey gelmesini bekler durumda olacaklardır.

Tüm thread'lerin bitmesini beklemeniz gerekiyor ise bir adet TCountDownEvent create eder ve ilk değerini thread sayısı olarak verirsiniz. Biten her thread TCountDownEvent nesnesinin Signal metodunu çağırır ve siz de uygulamanızın ana thread'inde; TCountDownEvent'i bekleyebilirsiniz. Ama benim bahsettiğim thread'in sonlanması. Yoksa, verilen işi yapıp işi bitirmesi değil. Zaten ilgili thread'iniz içinde bir nevi döngü ile dönmelisiniz.

Bunlar tabii, Delphi'nin sunduğu imkanlar. Windows API ile sunulan bir çok başka imkan da mevcut. Umarım; ne demek istediğinizi okuyabildiğim kısa paragraflardan doğru anlayabilmiş ve yardımcı olabilmişimdir.

Sizin için bir örnek hazırladım. Umarım faydalı olur.


  Marman_ThreadQueue_Deneme.zip (Dosya Boyutu: 54,07 KB / İndirme Sayısı: 41)

[vedat35]

(12-12-2021, Saat: 14:47)mrmarman Adlı Kullanıcıdan Alıntı: Merhaba.

* Forumlardaki başlıklara bakarken aklıma geldi. Bir proje tasarlayarak merakımı gidermeye karar verdim. Thread içerikli ve eksiklerimi pekiştirmek amaçlı.
* En altta düşündüğüm ve çalışacağına inancımın tam olduğu bir çözüm de var ama konuyu özetlemiş olayım isterim :

* Örnek sorunsal çekirdeği vereyim : 
* 13 tane dosya indireceğiz. ( Özellikle 13 dedim bir tane 10'lu bir tane 3'lü thread beklemesi ile küsuratlı olsun istedim... )
* Ağ sağlığı için en fazla 10 thread aynı anda çalışsın istiyoruz. ( Bu bile fazla ama örnekleme kolay olsun diye böyle diyelim. Mantığı kurunca ekonomisi yapılır...)
* Yalnız bir sonraki sekans için 10 thread'in 10'unu da bitene kadar beklemeyelim, 10'dan bitenin yerine kalan 3'den bir tanesi hemen devreye girsin. Çünkü 10 thread içerisinde en yavaş olan thread hızı tüm bekleme süresini belirleyecektir bunu biliyoruz.
 

* Genel uygulamam şu şekildedir. 

• Bir ana proje MainThread çalışıyor.
  
• Bu ana projeye işleri üstelenecek bir Worker thread eklerim. Böylece ana proje dahili işlemlerde bloke olmuyor. Eventleri Syncronize vb. ile çalıştırıp haberleştiriyoruz vs.. 
  
• Ardışık olarak internetten bir sürü dosya indirmek istiyoruz. 
  
• Bunun için WaitForMultipleObject'i başarılı bir şekilde kullanıyoruz. Mesela sadece 10 tane dosya simültane indirsin, tümü bitince sonraki 10 tane dosya indirme işlemine başlasın şekilde yürüyoruz.
  

Aklıma gelen soru bu noktada... 

• Thread sayısını sınırlandırdığımda (10 tane Threadhandle'ini bir Array ile vererek) WaitForMultipleObject ile (TRUE ile hepsi bitince talimatıyla)  tüm dosyalar indiğinde WAIT_OBJECT_0 şeklinde bir sonuç ile işlem başarılı bir şekilde sonuçlanarak bir sonraki sekansa geçmek için sebebim oluyor.
  
• Eğer hepsi bitince talimatını FALSE yaparsam önce hangisi biterse onun index nosu ile ( WAIT_OBJECT_0+1, WAIT_OBJECT_0+5, vb. ) array içerisinden hangisinin işi önce bitmişse onun index nosunu alıyorum.
  

İşte sorum bu noktada...   

10 Thread'in 10'unu da beklemeden bitenin yerine yenisini başlatmak istersem hemen altında yeni bir WaitFor kurmak için nasıl bir strateji kurmak gerekli.

- Düşüncemde yer eden çözüm; Recursive bir yapı ile "waitformultipleobject" / waitfor all = "FALSE" olarak  ilkinde 10 diğerlerinde tek elemanlı alt alta dallanan bir single çalışan multiobject yapısı gibi bir beklenti.

- Recursive derken kendi içinde kendini çağıran ama ilk çağrıda maksimum thread sayısı iken içe dallandığında daima 1 elemanlı array ile bekleyen bir yapı.

* Bu noktada çözmem gereken sorunsal ise alta dallandıktan sonra yine bir WaitFor koyup devam etmekte olan, ARRAY içindeki tüm thread'lerin bittiğinden de emin olmak için nasıl bir yapı kuracağım...?! Sanki Array içinde kalan elemanları da SUB WaitFor için gönderip onların da waitfor'a dahil olmasını beklemek gibi bir garip düşünce de oluşmadı değil...

* Bu şekilde çözebileceğimi biliyorum ve inanıyorum ama kolları sıvamadan sorayım dedim.... Mantıklı mıdır yoksa alternarifi var mıdır ? 

* Başlığa yabancılık çekenler için anahtar kelimeleri vereyim :
* WaitForSingleObject 1 tread sonlanana kadar procedure bekletir ve bitince procedure kaldığı yerden devam eder.
* WaitForMultipleObjects  ( n ) adet thread bekler...  bWaitAll= TRUE ise tümü bitmediği sürece bekler procedure de bekler, bWaitAll = FALSE ise en az birisi biterse, bekleme işlemi sona erer ama diğer threadler çalışmaya devam ederken procedure yeniden yürümeye başlar.

Hocam yoğun thread kullanımında, bu threadler her zaman oluşturulabilir anlamı taşımaz, unable to create thread cevabı sistemden alabilirsiniz, thread create exception a düşer, uygulama kitlenmiş bir şekilde beklemeye geçer, bunun ne zaman olacağınıda bilemeyiz ancak olması olasıdır,
bu sebeple sırf socket read write blocking yapılıyorsa, win ortamında async soket kullanmayı tavsiye ederim, 
böylelikle event notify mantığıyla ve uygulama blocklanmadan çok daha az bellek kullanan performanslı bir uygulamaya ortaya çıkar,
örneğin 5k client bağlantısı için 5k thread yaratmak hem client hem server soket için pek mantıklı değildir çünkü.

https://github.com/fundamentalslib/funda...ce/Sockets

[mrmarman]

Üstadlar ayrı ayrı teşekkür ediyorum.
Güzel fikirler verdiğiniz için.

[vedat35]

(13-12-2021, Saat: 21:58)mrmarman Adlı Kullanıcıdan Alıntı: Üstadlar ayrı ayrı teşekkür ediyorum.
Güzel fikirler verdiğiniz için.

hocam baştacısın.

[Tuğrul HELVACI]

(13-12-2021, Saat: 21:58)mrmarman Adlı Kullanıcıdan Alıntı: Üstadlar ayrı ayrı teşekkür ediyorum.
Güzel fikirler verdiğiniz için.

Rica ederim. Kodda anlaşılmayan, yardımcı olabileceğim bir şey var mı üstad ?

[mrmarman]

Üstadım bu hafta yeni yayın teknolojileri üzeribe eğitim veriyorum. Dün ilk gün olduğundan cihazların kurulumu / sökümü çok vakit aldı.
Bugün daha uygun olacağım. Akşam deneme fırsatım olacak.

[mrmarman]

@Tuğrul HELVACI

Biraz önce örnek kodu inceledim. Zekice kurgulanmış tebrik ederim. 
Mantık ; CPU çekirdek sayısı x 2 adet thread dizisini peşinen oluşturup çalışır halde bırakarak, sırasıyla işleri bunlara atamak. Thread görevi bitirdiğinde de çalışır kalmasına izin vermek.
Böylece thread oluştur thread bitince yenisini oluştur vs. derdi artık yok hükmünde.

* Başlığı okuyanlar için aşağıdaki listede görülmesi gereken 1744 Id'li thread (3) kere görev almış, 12344 Id'li thread (2) defa görev almış.

Kod: (Select All)

2688 ThreadID başlatıldı
8408 ThreadID başlatıldı
11648 ThreadID başlatıldı
16760 ThreadID başlatıldı
9320 ThreadID başlatıldı
6136 ThreadID başlatıldı
1744 ThreadID başlatıldı
16940 ThreadID başlatıldı
16620 ThreadID başlatıldı
9360 ThreadID başlatıldı
11000 ThreadID başlatıldı
4892 ThreadID başlatıldı
8948 ThreadID başlatıldı
5512 ThreadID başlatıldı
12344 ThreadID başlatıldı
13684 ThreadID başlatıldı
***********************************
Random Value: 990857453, Worker Thread ID: 1744, Çalışma zamanı: 1799, Tek Sayı: 245449623536840529, Çift Sayı: 245449623041411802
Random Value: 1449733987, Worker Thread ID: 12344, Çalışma zamanı: 2601, Tek Sayı: 525432158990596036, Çift Sayı: 525432158265729042
Random Value: 1490167247, Worker Thread ID: 5512, Çalışma zamanı: 2820, Tek Sayı: 555149606752973376, Çift Sayı: 555149606007889752
Random Value: 736899495, Worker Thread ID: 1744, Çalışma zamanı: 1412, Tek Sayı: 135755216801263504, Çift Sayı: 135755216432813756
Random Value: 1352173282, Worker Thread ID: 1744, Çalışma zamanı: 2555, Tek Sayı: 457093146138662881, Çift Sayı: 457093146814749522
Random Value: 1800462294, Worker Thread ID: 12344, Çalışma zamanı: 3314, Tek Sayı: 810416118028935609, Çift Sayı: 810416118929166756
2688 ThreadID durduruldu
8408 ThreadID durduruldu
11648 ThreadID durduruldu
16760 ThreadID durduruldu
9320 ThreadID durduruldu
6136 ThreadID durduruldu
1744 ThreadID durduruldu
16940 ThreadID durduruldu
16620 ThreadID durduruldu
9360 ThreadID durduruldu
11000 ThreadID durduruldu
4892 ThreadID durduruldu
8948 ThreadID durduruldu
5512 ThreadID durduruldu
12344 ThreadID durduruldu
13684 ThreadID durduruldu

@vkamadan paylaştığınız örneğiniz de sorduğum thread limitleme konusuna tam cevap niteliğinde. Ancak bir kısıt var. Şöyle ki aşağıdaki şekilde kurduğumda tüm threadler peşinen create ediliyor. Yani aşağıdaki akışa bakılınca, bu döngü 10.000 olsaydı 10.000 tane create edip sonra bunları TSemaphore.Create ile belirlediğimiz sayı kadarlık gruplar halinde çalıştıracaktı anlamına geliyor. 

* Sonradan Embarcadero help ile incelediğimde "TSemaphore" kullanımının aslında benim istediğim şekilde thread sayısı sınırlamaktan öte / farklı olarak aslında threadler ile ortak / paylaşılan kaynaklara limitli sayılarda thread ile kontrollü ve güvenli olarak erişim yönetimi için kullanıldığını görüyorum. 

var
  ASemaphore : TSemaphore;

procedure TForm1.BitBtn4Click(Sender: TObject);
var
  i : Integer;
begin
  if NOT Assigned(ASemaphore)
    then ASemaphore := TSemaphore.Create(nil, 3, System.CPUCount*2, '');

  for i := 1 to 13 do
  begin
    TThread.CreateAnonymousThread(
      procedure
      var
       ThId : word;
      begin
       ThId := i;
       TThread.Queue( nil, procedure begin form1.Memo1.Lines.Add(format('Thread %.2d created and running (%s) ', [ThId, DateTimeToStr(now)])); end );

       WaitForSingleObject(ASemaphore.Handle, INFINITE);
       Sleep(3000);
       ASemaphore.Release(1);

       TThread.Queue( nil, procedure begin form1.Memo1.Lines.Add(format('Thread %.2d exit (%s)', [ThId, DateTimeToStr(now)])); end );
      end
    ).start;
    Application.ProcessMessages;
    sleep(60);
  end;

end;

Kod: (Select All)

Thread 01 created and running (14.12.2021 21:09:17)
Thread 02 created and running (14.12.2021 21:09:17)
Thread 03 created and running (14.12.2021 21:09:17)
Thread 04 created and running (14.12.2021 21:09:17)
Thread 05 created and running (14.12.2021 21:09:17)
Thread 06 created and running (14.12.2021 21:09:17)
Thread 07 created and running (14.12.2021 21:09:17)
Thread 08 created and running (14.12.2021 21:09:17)
Thread 09 created and running (14.12.2021 21:09:17)
Thread 10 created and running (14.12.2021 21:09:17)
Thread 11 created and running (14.12.2021 21:09:17)
Thread 12 created and running (14.12.2021 21:09:18)
Thread 13 created and running (14.12.2021 21:09:18)
Thread 01 exit (14.12.2021 21:09:20)
Thread 02 exit (14.12.2021 21:09:20)
Thread 03 exit (14.12.2021 21:09:20)
Thread 09 exit (14.12.2021 21:09:23)
Thread 13 exit (14.12.2021 21:09:23)
Thread 08 exit (14.12.2021 21:09:23)
Thread 07 exit (14.12.2021 21:09:26)
Thread 10 exit (14.12.2021 21:09:26)
Thread 11 exit (14.12.2021 21:09:26)
Thread 12 exit (14.12.2021 21:09:29)
Thread 05 exit (14.12.2021 21:09:29)
Thread 04 exit (14.12.2021 21:09:29)
Thread 06 exit (14.12.2021 21:09:32)

@vedat35  vallahi söylediğiniz yeni anladım desem. "5k client / 5k server derken" zaten sorumun amacı yüzlerce veya binlerce kaynak için thread sayısı dinamiğini sınırlamak. Yani örneğin yukarıdaki örnekte maksimum "cpucore * 2" sağlıklı limitten sadece 3'erli gruplar haline iş için thread açılsın sonra kapananın yerine yenisi veya yenileri açılsın ama 3'ü geçmesin gibisinden bir beklentim vardı.

@Tuğrul HELVACI  ise işi kökünden halledip thread limit kadar oluşturup işleri bunlara dağıtarak çözüm önerisinde bulunmuş.

Hani "Matrix" filminin vurucu diyaloglarından biri ki izleyenleriniz vardır. 
- Soru "kaşığı zihin gücüyle nasıl büktüğümü merak ediyorsun değil mi?... Cevap "Aslında kaşık yok."    


Genel olarak bu minvalde zihnim aydınlandı. 
* Sadece nesnel programcılık için faydalı olsa da generics tipler konusuna biraz soğuğum. Bunu TThreadList / TList vb. ile yapmaya çalışacağım. 
* TObjectList yerine TThreadList kullanınca MemoryLeak sorunu yaşadım onun üzerine çalışayım.  

Tekrar önerileriniz ve vakit ayırma inceliğiniz için teşekkürler.

[Tuğrul HELVACI]

(14-12-2021, Saat: 21:37)mrmarman Adlı Kullanıcıdan Alıntı: @Tuğrul HELVACI

Biraz önce örnek kodu inceledim. Zekice kurgulanmış tebrik ederim. 
Mantık ; CPU çekirdek sayısı x 2 adet thread dizisini peşinen oluşturup çalışır halde bırakarak, sırasıyla işleri bunlara atamak. Thread görevi bitirdiğinde de çalışır kalmasına izin vermek.
Böylece thread oluştur thread bitince yenisini oluştur vs. derdi artık yok hükmünde.

* Başlığı okuyanlar için aşağıdaki listede görülmesi gereken 1744 Id'li thread (3) kere görev almış, 12344 Id'li thread (2) defa görev almış.

Kod: (Select All)

2688 ThreadID başlatıldı
8408 ThreadID başlatıldı
11648 ThreadID başlatıldı
16760 ThreadID başlatıldı
9320 ThreadID başlatıldı
6136 ThreadID başlatıldı
1744 ThreadID başlatıldı
16940 ThreadID başlatıldı
16620 ThreadID başlatıldı
9360 ThreadID başlatıldı
11000 ThreadID başlatıldı
4892 ThreadID başlatıldı
8948 ThreadID başlatıldı
5512 ThreadID başlatıldı
12344 ThreadID başlatıldı
13684 ThreadID başlatıldı
***********************************
Random Value: 990857453, Worker Thread ID: 1744, Çalışma zamanı: 1799, Tek Sayı: 245449623536840529, Çift Sayı: 245449623041411802
Random Value: 1449733987, Worker Thread ID: 12344, Çalışma zamanı: 2601, Tek Sayı: 525432158990596036, Çift Sayı: 525432158265729042
Random Value: 1490167247, Worker Thread ID: 5512, Çalışma zamanı: 2820, Tek Sayı: 555149606752973376, Çift Sayı: 555149606007889752
Random Value: 736899495, Worker Thread ID: 1744, Çalışma zamanı: 1412, Tek Sayı: 135755216801263504, Çift Sayı: 135755216432813756
Random Value: 1352173282, Worker Thread ID: 1744, Çalışma zamanı: 2555, Tek Sayı: 457093146138662881, Çift Sayı: 457093146814749522
Random Value: 1800462294, Worker Thread ID: 12344, Çalışma zamanı: 3314, Tek Sayı: 810416118028935609, Çift Sayı: 810416118929166756
2688 ThreadID durduruldu
8408 ThreadID durduruldu
11648 ThreadID durduruldu
16760 ThreadID durduruldu
9320 ThreadID durduruldu
6136 ThreadID durduruldu
1744 ThreadID durduruldu
16940 ThreadID durduruldu
16620 ThreadID durduruldu
9360 ThreadID durduruldu
11000 ThreadID durduruldu
4892 ThreadID durduruldu
8948 ThreadID durduruldu
5512 ThreadID durduruldu
12344 ThreadID durduruldu
13684 ThreadID durduruldu

@vkamadan paylaştığınız örneğiniz de sorduğum thread limitleme konusuna tam cevap niteliğinde. Ancak bir kısıt var. Şöyle ki aşağıdaki şekilde kurduğumda tüm threadler peşinen create ediliyor. Yani aşağıdaki akışa bakılınca, bu döngü 10.000 olsaydı 10.000 tane create edip sonra bunları TSemaphore.Create ile belirlediğimiz sayı kadarlık gruplar halinde çalıştıracaktı anlamına geliyor. 

* Sonradan Embarcadero help ile incelediğimde "TSemaphore" kullanımının aslında benim istediğim şekilde thread sayısı sınırlamaktan öte / farklı olarak aslında threadler ile ortak / paylaşılan kaynaklara limitli sayılarda thread ile kontrollü ve güvenli olarak erişim yönetimi için kullanıldığını görüyorum. 

var
  ASemaphore : TSemaphore;

procedure TForm1.BitBtn4Click(Sender: TObject);
var
  i : Integer;
begin
  if NOT Assigned(ASemaphore)
    then ASemaphore := TSemaphore.Create(nil, 3, System.CPUCount*2, '');

  for i := 1 to 13 do
  begin
    TThread.CreateAnonymousThread(
      procedure
      var
       ThId : word;
      begin
       ThId := i;
       TThread.Queue( nil, procedure begin form1.Memo1.Lines.Add(format('Thread %.2d created and running (%s) ', [ThId, DateTimeToStr(now)])); end );

       WaitForSingleObject(ASemaphore.Handle, INFINITE);
       Sleep(3000);
       ASemaphore.Release(1);

       TThread.Queue( nil, procedure begin form1.Memo1.Lines.Add(format('Thread %.2d exit (%s)', [ThId, DateTimeToStr(now)])); end );
      end
    ).start;
    Application.ProcessMessages;
    sleep(60);
  end;

end;

Kod: (Select All)

Thread 01 created and running (14.12.2021 21:09:17)
Thread 02 created and running (14.12.2021 21:09:17)
Thread 03 created and running (14.12.2021 21:09:17)
Thread 04 created and running (14.12.2021 21:09:17)
Thread 05 created and running (14.12.2021 21:09:17)
Thread 06 created and running (14.12.2021 21:09:17)
Thread 07 created and running (14.12.2021 21:09:17)
Thread 08 created and running (14.12.2021 21:09:17)
Thread 09 created and running (14.12.2021 21:09:17)
Thread 10 created and running (14.12.2021 21:09:17)
Thread 11 created and running (14.12.2021 21:09:17)
Thread 12 created and running (14.12.2021 21:09:18)
Thread 13 created and running (14.12.2021 21:09:18)
Thread 01 exit (14.12.2021 21:09:20)
Thread 02 exit (14.12.2021 21:09:20)
Thread 03 exit (14.12.2021 21:09:20)
Thread 09 exit (14.12.2021 21:09:23)
Thread 13 exit (14.12.2021 21:09:23)
Thread 08 exit (14.12.2021 21:09:23)
Thread 07 exit (14.12.2021 21:09:26)
Thread 10 exit (14.12.2021 21:09:26)
Thread 11 exit (14.12.2021 21:09:26)
Thread 12 exit (14.12.2021 21:09:29)
Thread 05 exit (14.12.2021 21:09:29)
Thread 04 exit (14.12.2021 21:09:29)
Thread 06 exit (14.12.2021 21:09:32)

@vedat35  vallahi söylediğiniz yeni anladım desem. "5k client / 5k server derken" zaten sorumun amacı yüzlerce veya binlerce kaynak için thread sayısı dinamiğini sınırlamak. Yani örneğin yukarıdaki örnekte maksimum "cpucore * 2" sağlıklı limitten sadece 3'erli gruplar haline iş için thread açılsın sonra kapananın yerine yenisi veya yenileri açılsın ama 3'ü geçmesin gibisinden bir beklentim vardı.

@Tuğrul HELVACI  ise işi kökünden halledip thread limit kadar oluşturup işleri bunlara dağıtarak çözüm önerisinde bulunmuş.

Hani "Matrix" filminin vurucu diyaloglarından biri ki izleyenleriniz vardır. 
- Soru "kaşığı zihin gücüyle nasıl büktüğümü merak ediyorsun değil mi?... Cevap "Aslında kaşık yok."    


Genel olarak bu minvalde zihnim aydınlandı. 
* Sadece nesnel programcılık için faydalı olsa da generics tipler konusuna biraz soğuğum. Bunu TThreadList / TList vb. ile yapmaya çalışacağım. 
* TObjectList yerine TThreadList kullanınca MemoryLeak sorunu yaşadım onun üzerine çalışayım.  

Tekrar önerileriniz ve vakit ayırma inceliğiniz için teşekkürler.

Üstad tekrar merhaba, memory leak benim gözümden kaçan bir kullanımdan kaynaklanmış. Thread'leri başlatmadan kuyruğa worker nesneleri eklersek ve thread'leri başlatmadan programdan çıkarsak; o nesneleri alıp işleyecek ve işi bittikten sonra yok edecek bir thread olmadığı için nesneler hafızada kalıyorlardı.

Bu ve buna benzer bir kaç minik sorunu düzeltip, kod içine bir kaç çok faydalı açıklama ve kullanım örnekleri ekledim.

Şiddetle yeni halini indirmenizi önerebilirim 


  Marman_ThreadQueue_Deneme_2.zip (Dosya Boyutu: 57,16 KB / İndirme Sayısı: 53)