[es] - Konkurentni Array?

mmix
Miljan Mitrović
Chief Software Architect
Financial Intelligence Software Ltd.
Passau, Deutschland

SuperModerator
Član broj: 17944
Poruke: 6050

+4631 Profil

Konkurentni Array?

^{14.10.2011. u 19:56 - pre 165 meseci}

Imam multihreaded proces, pokrenut kroz Parallel i on crunchuje nekih milion i sitno taskova. Rezultati tih taskova treba da idu u neki "niz" poredjan po istom indeksu kao source set (imam index u okviru taska, to nije problem) i mroam da imam neki IEnumerabe-like pristup tom rezultatu sortiran po indeksu.

Problem je sto ne bih da tim punjenjem usporavam taskove vise negos to je neophodno, problem 2 je sto ne daju svi taskovi rezultat (ok je da result[index] bude null) i sto raskovi traju arzliciti period vremena sto znaci da order(index) nije isti na ulazu u paralel i na izlazu iz taskova. Kad se ceo proces zavrsi, poslednji index "zatvara" strukturu i ta struktura postaje read-only (nema menjanja ni samih elemeanta ni strukture) i sluzi za storage i dalje crunchovanje u drugom multithreaded ciklusu koji je izdvojen.

Koliko vidim nijedna struktura u System.Collection.* ne radi to sto ja hocu, ponajmanje u Concurrency podspace-u jer algoritam nije producer/consumer vrste. Ako uzmem SortedDictionary<K=typeof(index), T> onda imam problem sto su svi upiti u dictionary ukljucujuci i njegov IEnumerable<KeyValuePair<K,T>> u rangu O(log n) sto ce mi ubiti pperformanse narednog ciklusa. U osnovi najidealnije resenje bi mi bio auto-grow array sa sync podrskom. postoji li uopste tako nesto ili mroam da pisem?

Sloba je za 12 godina promenio antropološki kod srpskog naroda. On je od jednog
naroda koji je bio veseo, pomalo površan, od jednog naroda koji je bio znatiželjan, koji
je voleo da vidi, da putuje, da upozna,
od naroda koji je bio kosmopolitski napravio narod koji je namršten, mrzovoljan,
sumnjicav, zaplašen, narod koji se stalno nešto žali, kome je stalno neko kriv - Z.Đinđić

Odgovor na temu

Mihajlo Cvetanović
Beograd

Član broj: 37636
Poruke: 1263

+96 Profil

Re: Konkurentni Array?

^{15.10.2011. u 15:48 - pre 165 meseci}

Šta fali nizu od milion elemenata? O(1)

Uzgred, ako je traženi kontejner samo međurezultat čitavog procesa, onda da li je zaista važno da elementi budu poređani?

Odgovor na temu

mmix
Miljan Mitrović
Chief Software Architect
Financial Intelligence Software Ltd.
Passau, Deutschland

SuperModerator
Član broj: 17944
Poruke: 6050

+4631 Profil

Re: Konkurentni Array?

^{15.10.2011. u 17:33 - pre 165 meseci}

Sinhronizacija je problem, narocito kasnije. Drugo moram da odrzim null-ove tamo gde je prvi proces zakucao jer je i to faktor u finalnoj agregaciji zajedno sa polozajem na kojem je pukao (spacing i grupacije nullova im govore o grupama koje ispadaju van modela). Trece, milion je sada ;) kroz godinu dve bice par miliona (tj bar se oni nadaju) tako da ne mogu bas da zakucam upper limit niti mi se svidja ideja da alociram giga rama na lookup niz ako ulazni set nije toliki.

Array sam i uradio sada, samo sto sam parcelisao u dva nivoa gde svaki set ima 500 elementata a primarni lookup drzi 200 setova sa inkrementom od 100, cisto da smanjim prealokaciju. Zakljucavanje celog niza sam smanjio samo na kriticne write operacije. U proncipu write one/read many, sve read/write operacije su O(1) sa periodicnim blokadama za resize.

Odgovor na temu

Mihajlo Cvetanović
Beograd

Član broj: 37636
Poruke: 1263

+96 Profil

Re: Konkurentni Array?

^{15.10.2011. u 19:12 - pre 165 meseci}

Ne znam da li sam baš sve dobro shvatio, ali.. Izlazni element prvog dela procesa može biti kombinacija izlaznog podatka i ulaznog indeksa. Pošto se zna indeks možemo da imamo i null kao izlaz. Izlazni kontejner može biti po jedna obična nesinhronizovana lista za svaku radnu nit. Jedino što je sinhronizovano je trenutni broj završenih izlaznih elemenata. Kad ovaj trenutni broj dostigne ukupan broj onda imaš sve što ti treba, samo raštrkano u nekoliko listi. Ako je problem raštrkanost onda lepo sredi podatke u neki novi kontejner po izboru i nastavi dalje sa drugim delom procesa. Ostale niti više nemaju šta da obrađuju pa su liste konačne po logici stvari, i zato i nema potrebe za sinhronizacijom listi.

Odgovor na temu

mmix
Miljan Mitrović
Chief Software Architect
Financial Intelligence Software Ltd.
Passau, Deutschland

SuperModerator
Član broj: 17944
Poruke: 6050

+4631 Profil

Re: Konkurentni Array?

^{15.10.2011. u 21:45 - pre 165 meseci}

Ok, ja sam verovatno kriv jer nisam bas dobro objasnio. Kako indeks napreduje tako se procesiranje komplikuje jer je result[index] = f(input[index], index, set[result[p], 0<p<index]), set nije mnogo veliki, kad dodje do milion naraste na nakih 1000+ upita u prethodne rezultate, sto u svakom slucaju znaci da moram da syncujem result set i ne mogu da particionisem rezultat kao sto mogu input. Minimum moram imati lock(result[index]) medjutim jedini nacin na koji to uspevam je da mi je ceo niz prealociran sto mi se nimalo ne svidja. Samim tim sto moram u resize u trenutku dodavanja moram da zakljucam master lock jer mogu dva upisa da odu preko trenutnog limita (sta vise to je realna situacija jer je resize skup). ovo u internalSetElement, lock (_mainStorageWriteLock)

I sebe samog sam uterao u bug jer nisam, u magnovenju, provalio da set oko indexa mozda jos nije uradjen

i dap osto podrzavam null ne mogu nid a znam da li jeste ili nije, to sad pokusavam da resim najefikasnije

Code (csharp):

    public class ExpandableThreadableArray<T> : IList<T>, IList, ICollection<T>, ICollection, IEnumerable<T>, IEnumerable, INotifyCollectionChanged

    {

        // virtual alocation storage implementation, max initial size = 100,000 before expensive resize of main partition, further increments in 50,000

        private const int BLOCKINITIAL = 200;

        private const int BLOCKINCREMENT = 100;

        private const int PAGESIZE = 500;

        private volatile int _arrayLength;

        private volatile bool _readonly;

        private volatile int _storageNumOfPages = BLOCKINITIAL;

        private T[][] _storage = new T[BLOCKINITIAL][];

        private readonly object[] _locks = new object[PAGESIZE];

        private readonly object _mainStorageWriteLock = new object();

        public ExpandableThreadableArray ()

          {

            // init locks

            for (int i = 0; i < PAGESIZE; i++) _locks[i] = new object();

            // initi storage

            for (int i = 0; i < BLOCKINITIAL; i++) _storage[i] = new T[PAGESIZE];

        }

        #region internal synced operations on storage

        private int internalSetElement(int index, T element)

        {

            if (_readonly) throw new NotSupportedException("Array is in readonly mode");

            // determine page number and index into page

            int pageIndex = index / PAGESIZE;

            int indexIntoPage = index % PAGESIZE;

            // if page requested is above and beyond its time to expand, pricy but needed, master lock fence is needed here

            // no one can get past this point unless current _storage has enough pages since _storage ref can change during lock

            lock (_mainStorageWriteLock)

            {

                if (pageIndex >= _storageNumOfPages) internalResizeStorage(pageIndex + 1);

                // obtain lock and set element

                lock (_locks[indexIntoPage]) _storage[pageIndex][indexIntoPage] = element;

                if (index >= _arrayLength) _arrayLength = index + 1;

            }

            return index;

        }

        private T internalGetElement(int index)

        {

            if (index >= _arrayLength) throw new InvalidOperationException("Requested index is out of array bounds");

            // determine page number and index into page

            int pageNumber = index / PAGESIZE;

            int indexIntoPage = index % PAGESIZE;

            // obtain lock and get element, , if array is readonly skip the locks

            if (_readonly) return _storage[pageNumber][indexIntoPage];

            else lock (_locks[indexIntoPage]) return _storage[pageNumber][indexIntoPage];

        }

        private int internalGetCount()

        {

            // volatile so no need to lock single op

            return _arrayLength;

        }

        private void internalClearArray()

        {

            // block everything and clear up the array, dont resize

            lock (_mainStorageWriteLock)

            {

                _readonly = false;

                for (int i = 0; i < _arrayLength; i++) internalSetElement(i, default(T)); // prevent live instances from remaining in the array

                _arrayLength = 0;

            }

        }

        private int internalAppendElement(T element)

        {

            // we must unfortunately lock entire storage for this as arraySize is tied to it

            lock(_mainStorageWriteLock) return internalSetElement(_arrayLength, element);

        }

        private void internalResizeStorage(int newMinimalNumberOfPages)

        {

            // everyone is already bloked for writing here, so we can proceed;

            // how many pages do we need

            int deltaPages = newMinimalNumberOfPages - _storageNumOfPages;

            int newIncrements = (deltaPages +  BLOCKINCREMENT - 1) / BLOCKINCREMENT;

            int newStoragePagesCount = _storageNumOfPages + newIncrements * BLOCKINCREMENT;

            // resize lookup array and preallocate pages

            Array.Resize(ref _storage, newStoragePagesCount);

            for (int i = _storageNumOfPages; i < newStoragePagesCount; i++) _storage[i] = new T[PAGESIZE];

            // udpate new storeage size and move on

            _storageNumOfPages = newStoragePagesCount;

        }

... implementacija interfejsa preko internal metoda ....

Odgovor na temu