Kicsit átheggesztettem a kódodat, mert úgy vettem észre, hogy az egyik szorulást az okozza, hogy a MessageQueue.Send() kicsit tovább tart, illetve nálad a Receive/Send/Receive/Send iterációk lineárisan vannak, emiatt a Receive-eknek meg kell várni a hosszabb Send-et, pedig a következő Receive már elindítható lenne, amig az előző Send tart.
Emiatt lehet kicsit gyorsítani, hogy ha a Receive lefutott, akkor azonnal indítani kell a következőt aszinkron módon, és csak utána hívni a Send-et.
Az alábbi kód nálam 2000 msg/s felett produkál (mondjuk nálam az eredeti kód is 300-400 körül futott, de monsjuk akkor is 5-6-szoros sebességnövekedés volt)
Bocs a retek kódért (tuple meg ilyenek), de most nem az igényeskedés volt a lényeg ;)

…
Console.WriteLine(“Starting to move data from source queue to destination queue”);

var sp = Stopwatch.StartNew();

using (var msmqTx = new MessageQueueTransaction())
{
var finishedEvent = new SemaphoreSlim(0, 4);

// Nálam négy kérést betárazva már elérte a maximumot 2 magos gépen.
q1.BeginReceive(TimeSpan.Zero, new Tuple(q1, q2, msmqTx, finishedEvent), ProcessReceive);
q1.BeginReceive(TimeSpan.Zero, new Tuple(q1, q2, msmqTx, finishedEvent), ProcessReceive);
q1.BeginReceive(TimeSpan.Zero, new Tuple(q1, q2, msmqTx, finishedEvent), ProcessReceive);
q1.BeginReceive(TimeSpan.Zero, new Tuple(q1, q2, msmqTx, finishedEvent), ProcessReceive);

finishedEvent.Wait();
} // using msmqTx
Console.WriteLine(“Duration: {0}ms, throughput: {1:F0} messages/s”, sp.ElapsedMilliseconds, 1000.0 * NumberOfTests / sp.ElapsedMilliseconds);
} // Main()

static void ProcessReceive(IAsyncResult ar)
{
var state = ar.AsyncState as Tuple;

try
{
// átvesszük az üzenetet
var message = state.Item1.EndReceive(ar);

// Mielőtt blokkolnánk a Send-del, elkezdjük a következő olvasást
state.Item1.BeginReceive(TimeSpan.Zero, state, ProcessReceive);

// Most már blokkolhat a Send
state.Item2.Send(message, state.Item3);
}
catch (MessageQueueException e)
{
Console.WriteLine(“Finished”);
state.Item4.Release();
} // catch()
} // ProcessReceive()

ja, és a szemafort ki kell cserélni CountdownEvent-re.

huhh, tényleg, a tranzakciókat elegánsan kihagytam. A gond az, hogy a BeginReceive-nek nincs is olyan változata, amelyik tud tranzakciókat. Ezért csináltam egy másik megoldást, de ez már közel sem olyan gyors:

…
Console.WriteLine(“Starting to move data from source queue to destination queue”);

var sp = Stopwatch.StartNew();

var task = Task.Factory.StartNew(
() => BridgeSegment(
new Tuple(
q1, q2)));
task.Wait();

Console.WriteLine(“Duration: {0}ms, throughput: {1:F0} messages/s”, sp.ElapsedMilliseconds, 1000.0 * NumberOfTests / sp.ElapsedMilliseconds);
} // Main()

static void BridgeSegment(Tuple state)
{
try
{
using (var msmqTx = new MessageQueueTransaction())
{
msmqTx.Begin();
var message = state.Item1.Receive(TimeSpan.Zero, msmqTx);

// meg van a csomag, kezdje el a következő iterációt még a send előtt
Task.Factory.StartNew(() => BridgeSegment(state), TaskCreationOptions.AttachedToParent);

state.Item2.Send(message, msmqTx);
msmqTx.Commit();
} // using msmqTx
}
catch (MessageQueueException e)
{
} // catch
} // BridgeSegment()

Símán használva olyan 5-600 msg/sec-cel megy. Kis trükköt használva, ha beállítjuk, hogy gyorsan tudjon a thread pool szálakat üzembehelyezni, akkor feltornázható kb 1000-1100 msg/sec-ig:
(ez kell a program elejére:)

ThreadPool.SetMinThreads(50, 1000);

Ugye az a gond, hogy alapból kevés szállal indul a pool, és mind el kezd várni a Send-en (vagy a receive-n vagy a commiton) emiatt hiába van már elküldve a következő taszk, hogy az is vegye a csomagokat, csak várnak a taszkok a queue-ban (mivel a kevés meglévő szál vár a send-receive-commit-on). A te eredeti megoldásoddal is ez a helyzet, emiatt nem nő az eredmény a MaxDop emelésével, mivel a thread pool csak óvatosan emeli a szálak számát, ha nincs beállítva a SetMinThreads. Az eredeti megoldásod a SetMinThread-del használva felmegy 700-800-ig. Hogy miért nem megy fel ez is 1000 fölé, nem jöttem rá.

Az mindenesetre szomorú, hogy szálak tucatjai nélkül milyen lassú, sok szálat indítva (vagy sok taszkot betárazva) meg valahogy nem olyan elegáns az egész. Van 20 szál, és mind a be van állva..

Aztán meg arra is kíváncsi vagyok, hogy ha a sok szál közül az egyiken nem ér el a commit-ig, hanem rollback-el, akkor mennyire stabil a dolog a többi szálon :)

Csak enge zavar az ilyen:

for (…;…; i++ ) …

Az’é bizonyos szint fölött már nem divat ez; vagy ez csak amolyan szísárpos lazaság?

Egyáltalán, minek formázod a kódot, ha a fordító fomázás nélkül is megérti. Az ugye csak amolyan szépségápolás, jó a szemnek, meg könnyebben átlátja mindenki, mit akarsz.
Akkor miért i++, ha egyszer ++i-t akartál írni?

Késő is van, fáradt is vagyok; belédkötöttem.

for (…;…; ++i) …

vagy ezmég nincs meg az a bizonyos szint?

Soci, nem lehetne valami kodkezelot engedni a kommentekhez? Ez igy borzasztoan olvashatatlan, ha az ember tanulna is a beszelgetesbol… :s

++i: nem értek egyet a következők miatt: A for ciklusban az i++/++i esetében nem azt fejezem ki, hogy milyen értéke kell az i-nek (növelés előtti vagy utáni). Az i++-szal azt fejezem ki, hogy mi történjen az i-vel (pl növekedjen meg) Nem érdekel a kifejezés értéke (használja valamire a runtime? nem!). A kifejezés mellékhatása kell! Ezt pedig tökéletesen kifejezi az i++. Miből jön a for ciklus C-ben? Ebből:

int i = 0;
while (i < 20)
{
…
i = i + 1; /* legyen i eggyel nagyobb */
}

Ebből lett ez:
for (int i = 0; i < 20; i = i + 1)

De mivel Dennis Ritchie meg a barátja szerette tömörséget, kitalálták ezt:

i += 1; /* legyen i eggyel nagyobb */

És mivel ez még mindig nem volt elég, jött ez:

i++; /* legyen i eggyel nagyobb */

De mivel a programozásban gyakori ez:

i = 0;
while (…)
{
f(array[i])
i = i + 1;
}

Ha az i++ pontosan az i = i + 1 rövidítése lenne, akkor a tömörségre vágyó Ritchie nem tudta volna felírni ezt:

i = 0;
while (…)
{
f(array[i++]);
}

Ezért okosan kitalálták, hogy az i++ kifejezés értéke legyen az i eredeti értéke, és akkor majd milyen jól néz ki a fenti ciklus.

Más esetekben, mint például ebben:

(i = i + 1) < 20 ami amúgy
(i += 1) < 20 ami amúgy lehetne
i++ < 20, ha a fenti példa miatt nem trükköznének a kifejezés értékével.

Szóval mivel vannak olyan helyzetek, ahol az i++ “eredeti értelmezése” kell, kitalálták, hogy akkor a ++i majd azt jelenti, mint az i += 1, az i++ meg mehet a tömbindexes helyzetekbe.

Ezekben a helyzetekben tehát számit mi a kifejezés értéke. Más esetekben, ahol csak a mellékhatás érdekes, ott teljesen mindegy, hogy melyik formát használja az ember. És a for ciklus loop expression-je egy olyan hely, ahol csak a mellékhatás érdekes. Emiatt mindegy, hogy i++ vagy ++i. Szerintem.
(amúgy az i++ levezetése a fentiekben csak spekuláció, nem biztos, hogy így történt)