Hur tycker ni jag ska göra detta lib trådsäkert

På rak arm kan jag tänka mig tre sätt att lösa detta på:

1. Read + Write-lock

2. Write-lock + copy

3. Kö

Lösning 1 och 2 använder sig av ett låsobjekt, alternativt ett per lista (det är dyrt med många små låsobjekt så testa vad som är mest effektivt för dig)

Lösning 2 är bättre om du har fler reads än writes och går ut på att du bara låser vid uppdatering av listan:

lock {
    var listCopy = dict[guid].Concat(new [] { item });
    dict[guid] = listCopy;
}

Lösning 3 är att du publicerar meddelanden till ett separat enkeltrådat kösystem som hanterar tilläggning av data i dina listor. Det finns garanterat tredjepartsbibliotek som kan göra detta åt dig.

        private readonly IDictionary<Guid, List<Subscription>> subscriptions;
        private readonly IDictionary<string, List<Type>> userSubscriptions;

Det som skalar dåligt med tvåan är C#:s list-implementation. En omuterbar länkad lista (cons-list) är väldigt attraktiv här då du kan lägga till ett element med konstant tid och inte behöver kopiera listan. Det kanske går att använda den från F# men jag har aldrig provat.

edit:
Du kan göra så här:

private static readonly FSharpList<Subscription> Nil = FSharpList<Subscription>.Empty;
private FSharpList<Subscription> Subscriptions = Nil;

private readonly object myLock = new object();

public void AddOne(Subscription s) {
    lock (myLock) {
        this.Subscriptions = FSharpList<Subscription>.Cons(s, this.Subscriptions);
    }
}

public IEnumerable<Subscription> GetSubscriptionsLockFree() {
    return this.Subscriptions;
}

För att slippa synka nycklarna på userSubscriptions bör du använda en ConcurrentDictionary

List-typen kan gott vara omutbar, men referensen behöver inte vara det Se edit ovan.

Du skapar en ny lista helt enkelt där du lägger till de som ska finnas kvar

var newList = Nil;

foreach (sub in subscriptions[guid].Where(s => s.ConnectionId != connectionId)) {
    newList = FSharpList<Subscription>.Cons(sub, newList);
}

Med en bra implementation borde inte den länkade listan vara en flaskhals. Även en vanlig lista har dålig prestanda när man tar bort i mitten eftersom varje efterföljande element måste flyttas ett snäpp åt vänster.

RemoveAll gör antagligen precis samma sak som koden jag skrev för den länkade listan, dvs skapar en ny lista och lägger i de objekt som inte matchar prediktatet. En naivare implementation som gjorde Remove() om och om igen skulle ge O(N^2). Även RemoveAt är O(N) på grund av minneskopieringen som görs (som är O(N))

Det är sant att O(N) inte är jämförbart med O(N) men Array.Copy är en O(N)-funktion
http://msdn.microsoft.com/en-us/library/z50k9bft.aspx

Kul, berätta gärna vad det blir för resultat

Rupertoo hade en väldigt relevant fråga i sitt första inlägg: måste Subscribe, Unsubscribe och UnsubscribeConnection ta effekt direkt när de anropas eller är det OK med "eventual consistency".

Det går att få detta att skala i princip linjärt till rätt många CPU-kärnor (förutsatt att VM-en inte ställer till det) om "eventual consistency" är OK.

Blir man inte i sådana fall tvungen att synkronisera "journalen" mellan trådar? Om två klienter registerar sig samtidigt måste man låsa så att bägge inte försöker köra this.eventJournal.Add(event) samtidigt. Eller menar ni att man på något sätt bör köra en journal per tråd och vid behov sammanställa dessa?

Min fundering är hur man kan lägga till ett ärende i en kö utan att låsa på samlingen: Det kan hända att två klienter vill lägga till varsitt ärende på samma gång, och då behövs någon sorts synkronisering av kön för att säkerställa att de båda Add-operationerna inte krockar.
Tänker jag fel? Ge gärna ett kodexempel.

Hade inte läst källkoden innan jag skrev min kommentar, bara läst tråden, så en revidering: det kommer att skala med eventual consistency" men det kommer bara nära linjärt när man har väldigt många subscribers så Subscribe, Unsubscribe, UnsubscribeConnection och Handle är relativt dyra. Problemet är att man kommer behöva en räknare som måste delas mellan alla trådar då man måste veta i vilken ordning Subscribe, Unsubscribe & UnsubscribeConnection anropades i... Hade ordningen man anropar dessa inte spelat någon roll hade man sluppit detta.

Sedan kan det faktiskt vara så att det inte skalar alls just nu för vem vet hur väl de anrop som faktiskt görs i koden skalar, t.ex. skapa en ny GUID från flera trådar.

Men i stort kan man ha M*N länkade listor (där N endera är ett fix tal, ett par gånger större än antalet trådar alt. så är N exakt lika med antalet trådar men har aldrig använt ThreadLocal<T> i C# så vet inte om den saknar lås eller ej). M är i detta fall 3 då vi har 3 olika metoder.

Används ThreadLocal<T> så lägger man bara in ett entry i listan som innehåller argumenten till funktionen + det globala sekvensnumret. Det lås som behövs här kommer nästa enbart tas av samma tråd == väldigt billigt då den cache-line som låset ligger på kommer alltid vara i tillstånd modified av den lokala CPU-kärnan så operationen är helt lokal.

I fallet där man har fix antal hinkar har man ett lås per hink + att varje hink måste ligga på en egen cache-line (hur fixar man det i C#???). Tråd ID använd sedan som ett hash-värde som avgör vilken hink man lägger samma typ av entry som ovan.

Med frekvens F (som ska vara relativt låg, max 100 gånger i sekunden, helst mindre) så går man igenom alla listor, utför operationerna i ordning och sparar resultatet i datastrukturer som initierats med strukturer som bara läses av Handle. När man är klar använder man Interlocked.CompareAndExchange för att göra de nya strukturerna synliga för Handle. I.e. Handle kan alltid läsa helt utan lås.

Read/write lås fungerar mycket mer sällan än man tycker att det borde. Anledningen är att detta typ av lås typiskt kräver att även läsare gör en skrivning till minst en cache-line (på något sätt måste man markera att en läsare är inom den skyddare regionen). Om detta görs från flera trådar blir den operationen väldigt dyr då cache-line:en kommer vara i tillstånd modified på "fel" CPU-kärnan, vilket betyder att den kärna som äger cache-line:en måste skriva ner resultatet (så den går till tillstånd exklusive), den nya tråden läser resultatet (så cache-line:en går till shared) för att sedan skriva att den kliver in (måste nu invalideras på den andra kärnan och sättas till modified på denna). Allt detta kan lätt ta många 100-tals cykler!

Vet man exakt hur många trådar man har och det antalet aldrig ändras kan man göra ett väldigt skalbara read/write-lås genom att ha en mutex per tråd. Läsare låser bara "sin" mutex som då blir billigt då det är en lokal operation. Skrivare låser alla lås, så man garanterar att ingen annan är inom regionen vid skrivning (som då måste vara mycket mer sällsynt än läsning). I detta fall blir läsare från olika trådar helt oberoende av varandra så det skalar i princip linjärt.

Jag förstod hur du tänkte med journalerna men missförstod hur du menade att det skalade. Det fetmarkerade i citeringen visar informationen jag saknade.

ConcurrentQueue borde vara perfekt för den lösningen. Det kommer trots detta behövas ett lås när man ska sammanställa den effektiva listan av subscribers, dvs vid varje Handle(), om jag inte misstar mig.

Förresten, blir detta upplägg en typ av STM?

STM == Software Transactional Memory, men efter att ha läst artikeln i länken inser jag att det bara är journaldelen som är liknande.

STM är väldigt trevligt att jobba med, men ingen har lyckats göra en implementation som är speciellt snabb. HTM (Hardware Transaction Memory) har däremot börjat dyka upp i ett par CPUer, finns bl.a. i senaste IBM POWER och finns även i de Haswell CPUer som har TSX. Har pratat med ingenjörer på Intel om TSX och de sa ganska uppriktigt att TSX inte är den slutgiltiga lösningen, men upp till 4 CPUer fungerar det riktigt bra!

Har lekt en del med ConcurrentQueue i .Net och det verkar inte riktigt som att den är tänkt för användarfallet att man har consumers/producers som lägger in/tar ut data med väldigt hög frekvens. Såg kommenterar att det mer är tänkt som en "säker kö som du vill använda från flera trådar", i.e. poängen är funktionen att man kan använda den från flera trådar, inte att det skulle vara supereffektivt. Gjorde en egen ganska naiv kö-implementation baserad på Interlock.CompareAndExchange() som hade väldigt snarligt beteende/skalning (verkar vara en pekare per ända man kör CompareAndExchange på).

Man kan göra en mer skalbar ConcurrentQueue genom att ha N köer. Initialt har man en "vänster-pekare" med index 1 och en högerpekare med index 0. Index väljer ut en av de N köerna, som var och en endera använder någon lock-less insert/remove eller helt enkelt ett lås per hink.
Lägger man till till vänster gör man en fetch_and_add på index (som då blir 2 och man får tillbaka 1), detta ger 1 mod N, d.v.s hink 1. Nästa gång man lägger till får man en 2:a, så det hamnar i hink 2 mod N.

Lägger man i stället till åt höger gör man i stället fetch_and_sub, så nästa index blir -2 och hinken blir -1 mod N, etc. Plocka ut element gör man genom att köra index åt motsatt håll.

En sådan kö kommer i princip aldrig blocka av att man köar i ena änden och plockar av i andra (vanligaste use-case:et), givet ett tillräckligt stort värde på N.