Нишки

Накратко: Нишката е отделен ред на изпълнение в програмата; няколко нишки работят едновременно и делят паметта, затова общ брояч се пази с lock или Interlocked.

Проблем

Три отчета, един по един

Програмата сваля три месечни отчета от сървъра. Тук свалянето е пауза: Thread.Sleep(ms) спира програмата за толкова милисекунди. Thread е от System.Threading, който е включен по подразбиране.

Program.cs

static void Download(string file, int ms)
{
    Thread.Sleep(ms);
    Console.WriteLine($"Свалих {file}");
}

static void Main()
{
    Download("март.pdf", 600);
    Download("април.pdf", 200);
    Download("май.pdf", 400);
    Console.WriteLine("Готово");
}

Изход

Свалих март.pdf
Свалих април.pdf
Свалих май.pdf
Готово

Паузите са 0,6, 0,2 и 0,4 секунди, общо 1,2 секунди. През цялото време програмата само чака мрежата. Ако имаше бутон „Отказ“, той нямаше да реагира. А трите отчета не зависят един от друг и спокойно могат да се свалят едновременно.

Идеята

Процес и нишка

Когато пуснеш програмата, операционната система ѝ дава процес: своя памет, свои отворени файлове. Вътре в процеса работят една или повече нишки (threads). Нишката е поредица от команди, които се изпълняват една след друга. Досега всяка твоя програма имаше една нишка: тази, която изпълнява Main.

Нишките в един процес делят паметта си: виждат едни и същи обекти. Това ги прави удобни (лесно си предават данни) и опасни (могат да се объркат взаимно, ще го видим след малко).

• Паралелно значи наистина едновременно, на различни ядра на процесора.
• Конкурентно значи на едно ядро, като се редуват толкова бързо, че изглежда едновременно.

За програмиста разликата рядко има значение: и в двата случая не знаеш коя нишка кога ще продължи.

По-точно, за любопитните

• Браузърът Chrome пуска отделен процес за всеки раздел. Затова един увиснал сайт не спира останалите, но и затова браузърът ползва много памет.
• Нишки ползва всяка програма, която прави няколко неща наведнъж: играта пуска звука в една нишка, а логиката в друга; уеб сървърът обслужва всеки клиент в своя нишка.
• Цената: всяка нишка има свой стек и паметта за нея се заделя отделно. Хиляди нишки струват. Затова в истински програми се ползва басейн от нишки (последната идея в урока).

Виж го

Първата нишка

Класът Thread е нишка. На конструктора му подаваш какво да прави: делегат ThreadStart, метод без параметри, който не връща нищо. Най-кратко е с ламбда:

Thread worker = new Thread(() => Download("април.pdf", 200));
worker.Start();
Console.WriteLine("Main не чака");
worker.Join();
Console.WriteLine("Отчетът е свален");

Изход

Main не чака
Свалих април.pdf
Отчетът е свален

• Start() пуска новата нишка и се връща веднага. Main продължава, без да чака, затова „Main не чака“ се отпечатва първо.
• Join() кара Main да изчака нишката да свърши.

Download е методът от първата карта. В C# изключенията не са проверени, затова тук няма throws и нищо не трябва да се хваща: Main си остава static void Main().

По-точно, за любопитните

Ламбдата е най-краткият начин, но не единственият. В стария код ще срещнеш и другите три:

class ReportJob
{
    public void Run()
    {
        Console.WriteLine("2. метод на обект");
    }
}

static void PrintReport()
{
    Console.WriteLine("1. статичен метод по име");
}

static void Main()
{
    Thread a = new Thread(PrintReport);
    Thread b = new Thread(new ReportJob().Run);
    Thread c = new Thread(delegate ()
    {
        Console.WriteLine("3. анонимен метод");
    });
    Thread d = new Thread(() => Console.WriteLine("4. ламбда"));
    foreach (Thread t in new Thread[] { a, b, c, d })
    {
        t.Start();
        t.Join();
    }
}

Изход

1. статичен метод по име
2. метод на обект
3. анонимен метод
4. ламбда

Наследник на Thread в C# не може да има: класът е sealed (запечатан). Анонимни класове също няма, а старият им двойник е анонимният метод delegate () { ... }. Затова задачата винаги е отделена от нишката: делегат, който може да се даде и на басейн от нишки. Ако нишката трябва да получи аргумент, има и ParameterizedThreadStart: new Thread(file => ...) и Start("март.pdf"). Аргументът там е object, затова ламбда, която хваща променлива, е по-удобна.

Виж го

Три наведнъж

Сега трите отчета, всеки в своя нишка:

long start = Environment.TickCount64;
Thread march = new Thread(() => Download("март.pdf", 600));
Thread april = new Thread(() => Download("април.pdf", 200));
Thread may = new Thread(() => Download("май.pdf", 400));
march.Start();
april.Start();
may.Start();
march.Join();
april.Join();
may.Join();
long ms = Environment.TickCount64 - start;
Console.WriteLine($"Готово за под секунда: {ms < 1000}");

Изход

Свалих април.pdf
Свалих май.pdf
Свалих март.pdf
Готово за под секунда: True

Нишките тръгват в реда март, април, май, но свършват по реда на паузите: април (0,2 s), май (0,4 s), март (0,6 s). Цялото чакане е колкото най-дългото, 0,6 секунди вместо 1,2.

Реда, в който нишките стигат до WriteLine, решава операционната система. Тук паузите се различават достатъчно, затова редът е един и същ. Ако бяха еднакви, при всяко пускане можеше да е различен.Environment.TickCount64 е броят милисекунди, откакто компютърът е пуснат: разликата между две четения е колко време е минало.

Пробвай

Start или извикване

Какво ще отпечата програмата? (Name е името на нишката)

Thread.CurrentThread.Name = "Main";
ThreadStart job = () => Console.WriteLine($"Работи в: {Thread.CurrentThread.Name}");
Thread t = new Thread(job);
t.Name = "сваляне";
job();
t.Start();
t.Join();

Отговор

Работи в: Main
Работи в: сваляне

Работи в: сваляне
Работи в: Main

Работи в: сваляне
Работи в: сваляне

Провери

Идеята

Животът на нишката

Нишката минава през няколко състояния, а ThreadState казва в кое е:

Thread t = new Thread(() => Download("май.pdf", 500));
Console.WriteLine(t.ThreadState);
t.Start();
Thread.Sleep(100);
Console.WriteLine(t.ThreadState);
t.Join();
Console.WriteLine(t.ThreadState);
try
{
    t.Start();
}
catch (ThreadStateException)
{
    Console.WriteLine("Втори Start: ThreadStateException");
}

Изход

Unstarted
WaitSleepJoin
Свалих май.pdf
Stopped
Втори Start: ThreadStateException

Състояние	Кога
Unstarted	създадена, още не е пусната със Start()
Running	работи или е готова да работи
WaitSleepJoin	чака: ключалка, друга нишка (Join), или определено време (Sleep)
Stopped	методът ѝ е свършил

Нишката се пуска само веднъж. Втори Start() на същата нишка хвърля ThreadStateException. По-кратко, само „работи ли“, казва IsAlive.

По-точно, за любопитните

Още за Thread:

• Interrupt() моли нишката да спре. Ако тя спи или чака, Sleep/Join хвърля ThreadInterruptedException. В C# то не е проверено: ако нишката не го хване, програмата спира с грешка. Спирането отвън с Abort() в днешния .NET не работи.
• Name, ManagedThreadId: име и номер на нишката. Thread.CurrentThread дава нишката, в която се изпълнява кодът.
• Thread.Yield() подсказва, че нишката може да отстъпи процесора на друга. Рядко има нужда от него.

Проблем

Загубените посещения

Сайтът брои посещенията. Две нишки обслужват посетители и всяка вика Increment() 100 000 пъти:

Program.cs

class Counter
{
    private int count = 0;

    public void Increment()
    {
        count++;
    }

    public int Count => count;
}

static void Main()
{
    Counter visits = new Counter();
    ThreadStart work = () =>
    {
        for (int i = 0; i < 100_000; i++)
        {
            visits.Increment();
        }
    };
    Thread first = new Thread(work);
    Thread second = new Thread(work);
    first.Start();
    second.Start();
    first.Join();
    second.Join();
    Console.WriteLine($"Посещения: {visits.Count}");
}

Очакваш 200 000. Пет пускания:

Изход (всеки път различен)

Посещения: 146528
Посещения: 120201
Посещения: 178089
Посещения: 152070
Посещения: 153205

Всеки път по-малко и всеки път различно. Причината е, че count++ не е една стъпка, а три:

int value = count;   // 1. прочети
value = value + 1;   // 2. добави 1
count = value;       // 3. запиши

class Counter
{
    private int count = 5;
    private readonly object sync = new object();

    public void Increment()
    {
        count++;   // прочети, добави 1, запиши
    }

    public void SafeIncrement()
    {
        lock (sync)
        {
            count++;
        }
    }

    public int Count => count;
}

Стъпка по стъпка, с думи

1. count е 5. Нишките A и B викат Increment() едновременно.
2. A прочита count: 5.
3. B прочита count, преди A да е записала: също 5.
4. A добавя 1 в своето копие и записва 6.
5. B също записва 6. Две извиквания, а count е 6, не 7: едното добавяне е загубено.
6. Отначало, с lock: A взима ключалката, прочита 5 и записва 6.
7. B стига до lock (sync) и чака, докато A пусне ключалката.
8. B взима ключалката, прочита 6 и записва 7: нищо не е загубено.

Когато двете нишки прочетат една и съща стойност, преди някоя да е записала, едното добавяне се губи. Това се казва състезание (race condition). Дали ще стане, зависи от това кога операционната система сменя нишките, затова резултатът е различен при всяко пускане.

Идеята

lock: по една

Решението е тези три стъпки да ги прави само една нишка наведнъж. Блокът lock прави точно това:

class Counter
{
    private int count = 0;
    private readonly object sync = new object();

    public void Increment()
    {
        lock (sync)
        {
            count++;
        }
    }

    public int Count
    {
        get
        {
            lock (sync)
            {
                return count;
            }
        }
    }
}

Изход

Посещения: 200000

sync е обикновен обект, който служи само за ключ: всеки обект в C# може да е ключалка. Нишката я взима, когато влезе в lock (sync), и я пуска, когато излезе, дори при изключение. Другата нишка, която иска същия ключ, чака на входа. Сега резултатът е 200 000 при всяко пускане. И четенето на Count е в lock: така нишката, която чете, вижда последната записана стойност.

Ключът е частно поле, а не this: така никой код отвън не може да заключи същия обект и да те накара да чакаш. Дума synchronized пред метода, както в Java, в C# няма. Най-близкото е атрибутът [MethodImpl(MethodImplOptions.Synchronized)], но той заключва целия обект (this), затова не се препоръчва:

using System.Runtime.CompilerServices;

class Counter
{
    private int count = 0;

    [MethodImpl(MethodImplOptions.Synchronized)]
    public void Increment()
    {
        count++;
    }

    public int Count => count;
}

За брояч има и готов клас: Interlocked. Interlocked.Increment(ref count) прави трите стъпки наведнъж, без ключалка. ref подава самото поле, а не копие на стойността му:

class Counter
{
    private int count = 0;

    public void Increment()
    {
        Interlocked.Increment(ref count);
    }

    public int Count => count;
}

Изход

Посещения: 200000

Тук Count чете без защита, защото Main го чете чак след Join. Join гарантира, че вижда всичко, което нишката е записала.

По-точно, за любопитните

• Monitor.Wait и Monitor.PulseAll: нишка в lock блок може да пусне ключалката и да заспи, докато друга не я събуди. Така кухнята чака поръчка, а касата чака кухнята да вземе предишната (задачата „производител и потребител“):

class OrderBox
{
    private string order;
    private readonly object sync = new object();

    public void Put(string newOrder)
    {
        lock (sync)
        {
            while (order != null)
            {
                Monitor.Wait(sync);
            }
            order = newOrder;
            Monitor.PulseAll(sync);
        }
    }

    public string Take()
    {
        lock (sync)
        {
            while (order == null)
            {
                Monitor.Wait(sync);
            }
            string result = order;
            order = null;
            Monitor.PulseAll(sync);
            return result;
        }
    }
}

static void Main()
{
    OrderBox box = new OrderBox();
    Thread kitchen = new Thread(() =>
    {
        for (int i = 0; i < 3; i++)
        {
            Console.WriteLine($"Кухнята прие {box.Take()}");
        }
    });
    kitchen.Start();
    box.Put("пица");
    box.Put("салата");
    box.Put("супа");
    kitchen.Join();
}

Изход

Кухнята прие пица
Кухнята прие салата
Кухнята прие супа

Monitor.Wait е в while, а не в if: нишката може да се събуди и без причина, затова проверява условието наново. Същото готово прави BlockingCollection<string> с boundedCapacity: 1: Add чака, ако е пълна, а Take чака, ако е празна.

• Мъртва хватка (deadlock): две нишки чакат една друга завинаги. Офисът държи принтера и чака скенера, а рецепцията държи скенера и чака принтера:

object printer = new object();
object scanner = new object();

Thread office = new Thread(() =>
{
    lock (printer)
    {
        Thread.Sleep(100);
        lock (scanner) { Console.WriteLine("Офисът копира"); }
    }
});
Thread reception = new Thread(() =>
{
    lock (scanner)
    {
        Thread.Sleep(100);
        lock (printer) { Console.WriteLine("Рецепцията копира"); }
    }
});

Програмата увисва и никога не печата нищо. Начинът да го избегнеш е всички нишки да взимат ключалките в един и същ ред.

• Приоритети и фонови нишки: Priority подсказва колко е важна нишката, но операционната система може да не се съобрази, затова не разчитай на нея за правилността на програмата. IsBackground = true прави нишката фонова: програмата свършва, без да я чака:

Thread report = new Thread(() => Download("годишен.pdf", 100));
report.Priority = ThreadPriority.Lowest;   // Lowest, BelowNormal, Normal, AboveNormal, Highest
report.IsBackground = true;               // програмата не чака фоновите нишки
Console.WriteLine($"{report.Priority}, фонова: {report.IsBackground}");

Изход

Lowest, фонова: True

• Готови колекции за много нишки са в System.Collections.Concurrent: ConcurrentDictionary, ConcurrentQueue, BlockingCollection. Методите им правят проверката и промяната наведнъж, без твой lock.

Пробвай

Два банкомата

Двама души теглят по 80 € от една сметка със 100 € в един и същ момент. Понякога и двамата получават парите, а балансът става -60 €. На кой ред е грешката?

Избери реда с грешката.

1class Account2{3 private int balance = 100;45 public void Withdraw(int amount)6 {7 if (balance >= amount)8 {9 balance -= amount;10 }11 }1213 public int Balance => balance;14}

Провери

Идеята

Басейн от нишки и Task

Да правиш нишка за всяка задача е скъпо, а при хиляда задачи и опасно. Басейнът от нишки (thread pool) е няколко нишки, които си взимат задачи от опашка. В .NET той е един за цялата програма, ThreadPool, и сам решава колко нишки да има. Задача му дава Task.Run (от System.Threading.Tasks):

static int SizeOf(string file, int ms)
{
    Thread.Sleep(ms);
    return file.Length * 100;
}

string[] files = { "март.pdf", "април.pdf", "май.pdf", "юни.pdf" };
List<Task<int>> sizes = new List<Task<int>>();
foreach (string file in files)
{
    sizes.Add(Task.Run(() => SizeOf(file, 300)));
}
for (int i = 0; i < files.Length; i++)
{
    Console.WriteLine($"{files[i]}: {sizes[i].Result} KB");
}

Изход

март.pdf: 800 KB
април.pdf: 900 KB
май.pdf: 700 KB
юни.pdf: 700 KB

• Task.Run приема задача, която връща резултат, дава я на басейна и веднага връща Task<int>: обещание, че резултатът ще дойде (в Java това е Future).
• .Result чака задачата да свърши и връща резултата. Ако задачата е хвърлила изключение, .Result го хвърля, увито в AggregateException.
• Басейнът не се спира накрая: нишките му са фонови, затова програмата свършва, когато свърши Main. Задача, която още работи тогава, просто се прекъсва.

Колко задачи вървят наведнъж, решава басейнът. Ако искаш да ги ограничиш, както басейн от 2 нишки, Parallel.For пуска тялото за всяко i в басейна, най-много MaxDegreeOfParallelism наведнъж, и чака всички:

string[] files = { "март.pdf", "април.pdf", "май.pdf", "юни.pdf" };
int[] sizes = new int[files.Length];
ParallelOptions options = new ParallelOptions();
options.MaxDegreeOfParallelism = 2;
Parallel.For(0, files.Length, options, i =>
{
    sizes[i] = SizeOf(files[i], 300);
});
for (int i = 0; i < files.Length; i++)
{
    Console.WriteLine($"{files[i]}: {sizes[i]} KB");
}

Изход

март.pdf: 800 KB
април.pdf: 900 KB
май.pdf: 700 KB
юни.pdf: 700 KB

.Result чака, като държи нишката заета. В уеб модула ще видиш await, който чака същата задача, без да заема нишка.

По-точно, за любопитните

Ако задачата не връща нищо, Task.Run дава Task без тип. Тогава чакаш всички с Task.WaitAll:

List<Task> jobs = new List<Task>();
for (int i = 1; i <= 5; i++)
{
    string file = $"отчет-{i}.pdf";
    jobs.Add(Task.Run(() => Download(file, 100)));
}
Task.WaitAll(jobs.ToArray());
Console.WriteLine("Всички свършиха.");

Изход от едно пускане

Свалих отчет-2.pdf
Свалих отчет-1.pdf
Свалих отчет-4.pdf
Свалих отчет-5.pdf
Свалих отчет-3.pdf
Всички свършиха.

Петте отчета вървят едновременно (басейнът има достатъчно нишки), затова редът им се сменя от пускане на пускане. По-старият начин е ThreadPool.QueueUserWorkItem, но той не дава задача, която да изчакаш.

Пробвай

Кой резултат пръв?

„архив.zip“ се сваля 0,6 s, а „бележка.txt“ 0,1 s. Какво ще отпечата програмата?

Task<int> big = Task.Run(() => SizeOf("архив.zip", 600));
Task<int> small = Task.Run(() => SizeOf("бележка.txt", 100));
Console.WriteLine($"Голям: {big.Result}");
Console.WriteLine($"Малък: {small.Result}");

Отговор

Малък: 1100
Голям: 900

Зависи от това коя задача свърши първа

Голям: 900
Малък: 1100

Провери

Упражнение

Програми с нишки

Реши задачите в редактора си.

1. Паралелна сума. Масивът sales има 1 000 000 продажби, sales[i] = i % 100. Раздели го на 4 равни части и сумирай всяка в своя нишка. Всяка нишка пише в свое място в масив long[] partial, затова не трябва синхронизация. Ламбдата в for хваща самата променлива на цикъла, а не стойността ѝ (урокът „Ламбди“): копирай номера на частта в нова променлива преди ламбдата. Изчакай ги с Join и събери частите:

Изход

Части: [12375000, 12375000, 12375000, 12375000]
Общо: 49500000

2. Сваляне с басейн. Свали шест отчета (март.pdf, април.pdf, май.pdf, юни.pdf, юли.pdf, август.pdf) с Parallel.For, по 3 наведнъж. Ползвай SizeOf(file, 300) от картата за басейна. Отпечатай размерите по реда на файловете, сбора и дали всичко е станало за под 1,5 секунди (едно след друго би било 1,8 s):

Изход

март.pdf: 800 KB
април.pdf: 900 KB
май.pdf: 700 KB
юни.pdf: 700 KB
юли.pdf: 700 KB
август.pdf: 1000 KB
Общо: 4800 KB
Под 1,5 секунди: True

3. Брояч без загуби. Четири нишки, всяка с по 50 000 посещения, увеличават общ брояч. Напиши брояча така, че резултатът да е винаги 200 000. Опитай първо без защита и виж колко се губи, после поправи веднъж с lock и веднъж с Interlocked.

Изход

Посещения: 200000

Още задачи

• Тест за брояча. Напиши xUnit тест (урокът „Тестове“), който пуска 8 нишки по 100 000 увеличения и проверява 800 000 с Assert.Equal. Пусни го срещу брояча без защита няколко пъти, после срещу брояча с lock. Тест, който пада понякога, е типичният знак за състезание. С 10 000 на нишка в .NET нишката често свършва, преди следващата да е тръгнала, и тестът минава: състезанието е там, но не се вижда.
• Мъртвата хватка. Пусни примера с принтера и скенера от „По-точно“ (с Start и Join) и виж как увисва. Поправи го, като и двете нишки взимат първо принтера.

Тест

Провери се

1. Каква е разликата между t.Start() и job(), когато job е делегатът на нишката?

Няма разликаjob() пуска нова нишка, а Start() неStart() пуска делегата в нова нишка, а job() го изпълнява веднага в текущата

2. Защо Main вика Join() преди да отпечата резултата?

За да изчака нишката да свършиЗа да спре нишката веднагаЗа да стартира нишката

3. Две нишки правят по 100 000 пъти count++ върху общо поле без синхронизация. Какво ще е count накрая?

Винаги 100 000Често по-малко от 200 000, и различно при всяко пусканеВинаги 200 000

4. Какво гарантира lock (sync) { ... }?

Че кодът в блока е по-бързЧе нишките се изпълняват по реда на стартаЧе в даден момент само една нишка е вътре в блок, заключен със същия обект sync

5. Пускаш три задачи с Task.Run и пазиш Task<int> за всяка. Какво прави task.Result?

Пуска задачата нановоЧака задачата да свърши и връща резултата ѝВръща резултата веднага, дори да не е готов

Провери

Започни урока отначало

Свързани уроци

• Нужно отпреди: try, catch и finally · Списъци · Ламбди и делегати · Тестове с xUnit
• Продължава в: Мрежово програмиране · async и await · WebAPI: данни за други програми · Quick sort и лошият пивот
• Задачи по темата: Програми в истинския свят