Chủ đề trong C # – Phần 17: Wait and Pulse (2)

(Đăng ngày 07/12/2007) Một ứng dụng Wait / Pulse đơn giản là hàng đợi nhà sản xuất-người tiêu dùng – cấu trúc mà chúng tôi đã viết trước đó bằng cách sử dụng AutoResetEvent. Một nhà sản xuất xếp hàng đợi các tác vụ (thường là trên chuỗi chính), trong khi một hoặc nhiều người tiêu dùng đang chạy trên chuỗi công nhân chọn và thực hiện từng tác vụ một.

Nhà sản xuất / Người tiêu dùng Hàng đợi

Một ứng dụng Wait / Pulse đơn giản là hàng đợi của nhà sản xuất-người tiêu dùng – cấu trúc mà chúng tôi đã viết trước đó bằng cách sử dụng AutoResetEvent. Một nhà sản xuất xếp hàng đợi các tác vụ (thường là trên chuỗi chính), trong khi một hoặc nhiều người tiêu dùng đang chạy trên chuỗi công nhân chọn và thực hiện từng tác vụ một.

Trong ví dụ này, chúng tôi sẽ sử dụng một chuỗi để biểu diễn một nhiệm vụ. Hàng đợi nhiệm vụ của chúng ta sau đó trông như thế này:

Xếp hàng taskQ = new Queue();

Đăng ký ngay

Đăng ký ngay

Bởi vì hàng đợi sẽ được sử dụng trên nhiều luồng, chúng ta phải bọc tất cả các câu lệnh đọc hoặc ghi vào hàng đợi trong một khóa. Đây là cách chúng tôi sắp xếp một nhiệm vụ:

lock (tủ khóa) {
taskQ.Enqueue (“nhiệm vụ của tôi”);
Monitor.PulseAll (tủ khóa); // Chúng tôi đang thay đổi một điều kiện chặn
}

Bởi vì chúng tôi đang sửa đổi một điều kiện chặn tiềm năng, chúng tôi phải xung. Chúng tôi gọi là PulseAll hơn là Pulse vì chúng tôi sẽ cho phép nhiều người tiêu dùng. Nhiều hơn một chủ đề có thể đang đợi.

Chúng tôi muốn nhân viên chặn trong khi không có gì để làm, hay nói cách khác, khi không có mục nào trong hàng đợi. Do đó, điều kiện chặn của chúng tôi là taskQ.Count == 0. Đây là câu lệnh Wait thực hiện chính xác điều này:

lock (khóa)
while (taskQ.Count == 0) Monitor.Wait (khóa);
Bước tiếp theo là worker để dequeue nhiệm vụ và thực hiện nó:
lock (khóa)
while (taskQ.Count == 0) Monitor.Wait (locker);

nhiệm vụ chuỗi;
lock (khóa)
task = taskQ.Dequeue ();

Tuy nhiên, logic này không an toàn theo chuỗi: chúng tôi đã đưa ra quyết định hủy hàng dựa trên thông tin cũ – có được trong một tuyên bố khóa trước. Hãy xem xét điều gì sẽ xảy ra nếu chúng ta bắt đầu đồng thời hai chủ đề người tiêu dùng, với một mặt hàng đã có trong hàng đợi. Có thể cả hai luồng đều không vào vòng lặp while để chặn – cả hai đều nhìn thấy một mục duy nhất trên hàng đợi. Sau đó, cả hai đều cố gắng dequeue cùng một mặt hàng, tạo ra một ngoại lệ trong trường hợp thứ hai! Để khắc phục điều này, chúng tôi chỉ cần giữ khóa lâu hơn một chút – cho đến khi chúng tôi hoàn thành việc tương tác với hàng đợi:

nhiệm vụ chuỗi;
lock (tủ khóa) {
while (taskQ.Count == 0) Monitor.Wait (tủ khóa);
task = taskQ.Dequeue ();
}

(Chúng tôi không cần phải gọi Pulse sau khi giảm giá trị, vì không người tiêu dùng nào có thể bỏ chặn bởi có ít mặt hàng hơn trong hàng đợi).

Sau khi nhiệm vụ được hủy bỏ, không có yêu cầu nào khác để giữ khóa. Việc phát hành nó tại thời điểm này cho phép người tiêu dùng thực hiện một công việc có thể tốn thời gian mà không chặn các luồng khác không cần thiết.

Đây là chương trình hoàn chỉnh. Như với phiên bản AutoResetEvent, chúng tôi sắp xếp một nhiệm vụ null để báo hiệu người tiêu dùng thoát (sau khi hoàn thành bất kỳ tác vụ nào chưa hoàn thành). Bởi vì chúng tôi đang hỗ trợ nhiều người tiêu dùng, chúng tôi phải đặt một nhiệm vụ rỗng cho mỗi người tiêu dùng để đóng hoàn toàn hàng đợi:

Wait / Pulse Boilerplate # 2: Producer / Consumer Queue

sử dụng Hệ thống;
sử dụng System.Threading;
sử dụng System.Collections.Generic;

public class TaskQueue: IDisposable {
object locker = new object ();
Thread [] công nhân;
Xếp hàng taskQ = new Queue();

public TaskQueue (int workerCount) {worker
= new Thread [workerCount];

// Tạo và bắt đầu một luồng riêng cho từng worker
for (int i = 0; i <workerCount; i ++)
( worker [i] = new Thread (Consume)). Start ();
}

public void Dispose () {
// Yêu cầu một tác vụ rỗng cho mỗi công nhân để thực hiện mỗi lần thoát.
foreach (Công nhân luồng trong công nhân) EnqueueTask (null);
foreach (Công nhân luồng trong công nhân) worker.Join ();
}

public void EnqueueTask (string task) {
lock (locker) {
taskQ.Enqueue (task);
Monitor.PulseAll (tủ khóa);
}
}

void Consume () {
while (true) {
tác vụ chuỗi;
lock (tủ khóa) {
while (taskQ.Count == 0) Monitor.Wait (tủ khóa);
task = taskQ.Dequeue ();
}
if (task == null) return; // Điều này báo hiệu thoát khỏi
Console.Write (task);
Thread.Sleep (1000); // Mô phỏng tác vụ tốn thời gian
}
}
}

Đây là một phương thức Chính bắt đầu một hàng đợi tác vụ, chỉ định hai chuỗi người dùng đồng thời và sau đó xếp hàng mười nhiệm vụ sẽ được chia sẻ giữa hai người dùng:

static void Main () {
using (TaskQueue q = new TaskQueue (2)) {
for (int i = 0; i <10; i ++)
q.EnqueueTask (“Task” + i);

Console.WriteLine (“10 nhiệm vụ được xếp vào hàng”);
Console.WriteLine (“Đang đợi các tác vụ hoàn thành …”);
}
// Thoát khỏi câu lệnh using chạy phương thức TaskQueue’s Dispose, phương thức này
// tắt người tiêu dùng, sau khi tất cả các tác vụ chưa hoàn thành.
Console.WriteLine (“\ r \ nTất cả các tác vụ đã hoàn thành!”);
}

Phù hợp với mẫu thiết kế của chúng tôi, nếu chúng tôi loại bỏ PulseAll và thay thế Wait bằng khóa chuyển đổi, chúng tôi sẽ nhận được cùng một đầu ra.

Kinh tế xung

Hãy xem lại nhà sản xuất sắp xếp một nhiệm vụ:

lock (khóa) {
taskQ.Enqueue (task);
Monitor.PulseAll (tủ khóa);
}

Nói một cách chính xác, chúng tôi có thể tiết kiệm bằng cách chỉ xung khi có khả năng giải phóng một công nhân bị chặn:

lock (khóa) {
taskQ.Enqueue (task);
if (taskQ.Count <= worker.Length) Monitor.PulseAll (tủ khóa);
}

Tuy nhiên, chúng tôi sẽ tiết kiệm được rất ít, vì quá trình tạo xung thường diễn ra dưới một micro giây và không phải chịu chi phí nào đối với những người làm việc bận rộn – vì họ vẫn bỏ qua nó! Đó là một chính sách tốt với mã đa luồng để loại bỏ mọi logic không cần thiết: một lỗi không liên tục do một sai lầm ngớ ngẩn là một cái giá quá đắt để tiết kiệm một micro giây! Để chứng minh, đây là tất cả những gì cần thiết để giới thiệu một lỗi “công nhân bị mắc kẹt” không liên tục mà rất có thể sẽ trốn tránh thử nghiệm ban đầu (phát hiện sự khác biệt):

lock (khóa) {
taskQ.Enqueue (task);
if (taskQ.Count <worker.Length) Monitor.PulseAll (tủ khóa);
}

Pulsing bảo vệ chúng ta khỏi loại lỗi này một cách vô điều kiện.

Pulse hay PulseAll?

Ví dụ này đi kèm với tiềm năng kinh tế xung hơn nữa. Sau khi sắp xếp một nhiệm vụ, chúng ta có thể gọi Pulse thay vì PulseAll và không có gì sẽ phá vỡ.

Hãy tóm tắt lại sự khác biệt: với Pulse, tối đa một luồng có thể hoạt động (và kiểm tra lại điều kiện chặn vòng lặp while của nó); với PulseAll, tất cả các chuỗi đang chờ sẽ hoạt động (và kiểm tra lại điều kiện chặn của chúng). Nếu chúng tôi yêu cầu một nhiệm vụ duy nhất, chỉ một công nhân có thể xử lý nó, vì vậy chúng tôi chỉ cần đánh thức một công nhân bằng một Xung duy nhất. Nó giống như việc có một lớp học gồm những đứa trẻ đang ngủ – nếu chỉ có một cây kem thì chẳng ích gì để đánh thức tất cả chúng để xếp hàng mua nó!

Trong ví dụ của chúng tôi, chúng tôi chỉ bắt đầu hai chủ đề người tiêu dùng, vì vậy chúng tôi sẽ thu được rất ít. Nhưng nếu chúng tôi bắt đầu có mười người tiêu dùng, chúng tôi có thể có lợi đôi chút khi chọn Pulse thay vì PulseAll. Tuy nhiên, nó có nghĩa là nếu chúng tôi xếp hàng nhiều nhiệm vụ, chúng tôi sẽ cần phải Xung nhiều lần. Điều này có thể được thực hiện trong một câu lệnh khóa, như sau:

lock (khóa) {
taskQ.Enqueue (“task 1”);
taskQ.Enqueue (“nhiệm vụ 2”);
Monitor.Pulse (tủ khóa); // “Báo hiệu đến hai
Monitor.Pulse (locker); // luồng đang chờ.”
}

Giá của một Pulse quá ít là một công nhân bị mắc kẹt. Điều này thường sẽ biểu hiện dưới dạng một lỗi không liên tục, vì nó sẽ xuất hiện chỉ khi người tiêu dùng ở trạng thái Chờ đợi. Do đó, người ta có thể mở rộng câu châm ngôn trước đó “nếu nghi ngờ, Pulse”, thành “nếu nghi ngờ, PulseAll!”

Một ngoại lệ có thể xảy ra đối với quy tắc có thể phát sinh nếu việc đánh giá điều kiện chặn tốn nhiều thời gian một cách bất thường.

Sử dụng thời gian chờ

Đôi khi nó có thể không hợp lý hoặc không thể Xung bất cứ khi nào phát sinh điều kiện bỏ chặn. Một ví dụ có thể là nếu điều kiện chặn liên quan đến việc gọi một phương thức lấy thông tin từ việc truy vấn định kỳ cơ sở dữ liệu. Nếu độ trễ không phải là vấn đề, thì giải pháp rất đơn giản: người ta có thể chỉ định thời gian chờ khi gọi Wait, như sau:

lock (khóa) {
while (điều kiện khóa)
Monitor.Wait (khóa, hết giờ);

Điều này buộc điều kiện chặn phải được kiểm tra lại, tối thiểu, ở một khoảng thời gian đều đặn được chỉ định bởi thời gian chờ, cũng như ngay khi nhận được xung. Điều kiện chặn càng đơn giản, thời gian chờ càng nhỏ mà không gây ra sự kém hiệu quả.

Hệ thống tương tự hoạt động tốt như nhau nếu không có xung do lỗi trong chương trình! Có thể đáng giá thêm thời gian chờ cho tất cả các lệnh Chờ trong các chương trình mà đồng bộ hóa đặc biệt phức tạp – như một bản sao lưu cuối cùng cho các lỗi xung khó hiểu. Nó cũng cung cấp một mức độ miễn nhiễm lỗi nếu chương trình được sửa đổi sau đó bởi một người nào đó không có trong Pulse!

Cuộc đua và sự thừa nhận

Giả sử chúng ta muốn một tín hiệu cho một công nhân năm lần liên tiếp:

class Race {
static object locker = new object ();
bool tĩnh đi;

static void Main () {
new Thread (SaySomething) .Start ();

for (int i = 0; i <5; i ++) {
lock (locker) {go = true; Monitor.Pulse (tủ khóa); }
}
}

static void SaySomething () {
for (int i = 0; i <5; i ++) {
lock (locker) {
while (! go) Monitor.Wait (locker);
go = false;
}
Console.WriteLine (“Wassup?”);
}
}
}

Đầu ra mong đợi:

Sản lượng thực tế:

Chương trình này có sai sót: vòng lặp for trong luồng chính có thể tự do quay qua năm lần lặp lại của nó bất kỳ lúc nào nhân viên không giữ khóa. Có thể trước khi công nhân bắt đầu! Ví dụ Nhà sản xuất / Người tiêu dùng không gặp phải vấn đề này bởi vì nếu luồng chính đi trước nhân viên, mỗi yêu cầu sẽ đơn giản xếp hàng. Nhưng trong trường hợp này, chúng ta cần luồng chính chặn ở mỗi lần lặp lại nếu nhân viên vẫn bận với tác vụ trước đó.

Một giải pháp đơn giản là cho luồng chính đợi sau mỗi chu kỳ cho đến khi cờ đi được xóa bởi công nhân. Sau đó, điều này yêu cầu nhân viên gọi Pulse sau khi xóa cờ đi:

class Acknowledged {
static object locker = new object ();
bool tĩnh đi;

static void Main () {
new Thread (SaySomething) .Start ();

for (int i = 0; i <5; i ++) {
lock (locker) {go = true; Monitor.Pulse (tủ khóa); }
lock (khóa) {while (go) Monitor.Wait (khóa); }
}
}

static void SaySomething () {
for (int i = 0; i <5; i ++) {
lock (locker) {
while (! go) Monitor.Wait (locker);
go = false; Monitor.Pulse (tủ khóa); // Người làm việc phải Pulse
}
Console.WriteLine (“Wassup?”);
}
}
}

Một tính năng quan trọng của một chương trình như vậy là worker sẽ giải phóng khóa của nó trước khi thực hiện công việc có thể tốn thời gian của nó (điều này sẽ xảy ra ở vị trí mà chúng ta đang gọi là Console.WriteLine). Điều này đảm bảo kẻ chủ mưu không bị chặn quá mức trong khi nhân viên thực hiện nhiệm vụ mà nó đã được báo hiệu (và chỉ bị chặn nếu nhân viên đang bận với một nhiệm vụ trước đó).

Trong ví dụ này, chỉ một luồng (luồng chính) báo hiệu cho worker thực hiện một tác vụ. Nếu nhiều luồng để báo hiệu cho worker – sử dụng logic của phương pháp Main của chúng tôi – chúng tôi sẽ gỡ bỏ. Hai luồng báo hiệu có thể thực thi dòng mã sau theo trình tự:

lock (khóa) {go = true; Monitor.Pulse (tủ khóa); }

dẫn đến tín hiệu thứ hai bị mất nếu công nhân không thực hiện xong quá trình xử lý đầu tiên. Chúng tôi có thể làm cho thiết kế của mình trở nên mạnh mẽ trong trường hợp này bằng cách sử dụng một cặp cờ – cờ “sẵn sàng” cũng như cờ “đi”. Cờ “sẵn sàng” cho biết rằng công nhân có thể chấp nhận một nhiệm vụ mới; cờ “go” là một hướng dẫn để tiếp tục, như trước đây. Điều này tương tự với một ví dụ trước đó đã thực hiện điều tương tự bằng cách sử dụng hai AutoResetEvents, ngoại trừ khả năng mở rộng hơn. Đây là mẫu, được cấu trúc lại với các trường mẫu:

Wait / Pulse Boilerplate # 3: Tín hiệu hai chiều

public class Acknowledged {
object locker = new object ();
bool đã sẵn sàng;
bool đi;

public void NotifyWhenReady () {
lock (locker) {
// Chờ nếu nhân viên đã bận công việc trước đó
trong khi (! ready) Monitor.Wait (locker);
sẵn sàng = sai;
go = true;
Monitor.PulseAll (tủ khóa);
}
}

public void AcknowledgedWait () {
// Cho biết rằng chúng tôi đã sẵn sàng xử lý yêu cầu
khóa (locker) {ready = true; Monitor.Pulse (tủ khóa); }

lock (khóa) {
while (! go) Monitor.Wait (khóa); // Chờ tín hiệu “đi”
go = false; Monitor.PulseAll (tủ khóa); // Tín hiệu xác nhận
}

Console.WriteLine (“Wassup?”); // Thực hiện tác vụ
}
}

Để chứng minh, chúng tôi sẽ bắt đầu hai luồng đồng thời, mỗi luồng sẽ thông báo cho nhân viên năm lần. Trong khi đó, chuỗi chính sẽ đợi mười thông báo:

public class Test {
static Acknowledged a = new Acknowledged ();

static void Main () {
new Thread (Notify5) .Start (); // Chạy đồng thời hai
Thread mới (Notify5) .Start (); // thông báo …
Wait10 (); // … và một người phục vụ.
}

static void Notify5 () {
for (int i = 0; i <5; i ++)
a.NotifyWhenReady ();
}

static void Wait10 () {
for (int i = 0; i <10; i ++)
a.AcknowledgedWait ();
}
}

Trong phương pháp Thông báo, cờ sẵn sàng bị xóa trước khi thoát khỏi câu lệnh khóa. Điều này cực kỳ quan trọng: nó ngăn hai bộ thông báo phát tín hiệu tuần tự mà không cần kiểm tra lại cờ. Để đơn giản hơn, chúng tôi cũng đặt cờ đi và gọi PulseAll trong cùng một câu lệnh khóa – mặc dù chúng tôi cũng có thể đặt cặp câu lệnh này trong một khóa riêng biệt và không có gì bị phá vỡ.

Mô phỏng Xử lý Chờ

Bạn có thể nhận thấy một mẫu trong ví dụ trước: cả hai vòng lặp chờ đều có cấu trúc sau:

lock (khóa) {
while (! flag) Monitor.Wait (khóa);
cờ = sai;
…
}

nơi cờ được đặt thành true trong một chuỗi khác. Trên thực tế, điều này bắt chước một AutoResetEvent. Nếu chúng ta bỏ qua flag = false, thì chúng ta sẽ có một ManualResetEvent. Sử dụng trường số nguyên, Pulse and Wait cũng có thể được sử dụng để bắt chước Semaphore. Trên thực tế, Wait Handle duy nhất mà chúng ta không thể bắt chước với Pulse and Wait là Mutex, vì chức năng này được cung cấp bởi câu lệnh khóa.

Trong hầu hết các trường hợp, việc mô phỏng các phương thức tĩnh hoạt động trên nhiều Wait Handle là điều dễ dàng. Tương đương với việc gọi WaitAll trên nhiều EventWaitHandles không gì khác hơn là một điều kiện chặn kết hợp tất cả các cờ được sử dụng thay cho Wait Handles:

lock (khóa) {
while (! flag1 &&! flag2 &&! flag3 …) Monitor.Wait (locker);

Điều này có thể đặc biệt hữu ích vì WaitAll trong hầu hết các trường hợp không thể sử dụng được do các sự cố kế thừa COM. Mô phỏng WaitAny chỉ đơn giản là việc thay thế toán tử && bằng || nhà điều hành.

SignalAndWait phức tạp hơn. Nhớ lại rằng phương pháp này báo hiệu một xử lý trong khi chờ xử lý khác trong một hoạt động nguyên tử. Chúng ta có một tình huống tương tự như một giao dịch cơ sở dữ liệu phân tán – chúng ta cần một cam kết hai giai đoạn! Giả sử chúng ta muốn báo hiệu cho flagA trong khi chờ flagB, chúng ta sẽ phải chia mỗi cờ thành hai, dẫn đến mã có thể trông giống như sau:

lock (khóa) {
flagAphase1 = true;
Monitor.Pulse (tủ khóa);
while (! flagBphase1) Monitor.Wait (khóa);

flagAphase2 = true;
Monitor.Pulse (tủ khóa);
while (! flagBphase2) Monitor.Wait (khóa);
}

có lẽ với logic “quay lại” bổ sung để rút lại flagAphase1 nếu câu lệnh Wait đầu tiên ném ra một ngoại lệ do bị gián đoạn hoặc bị hủy bỏ. Đây là một tình huống mà Xử lý Chờ dễ dàng hơn! Tuy nhiên, tín hiệu nguyên tử thực sự và sự chờ đợi thực sự là một yêu cầu bất thường.

Chờ Điểm hẹn

Cũng giống như WaitHandle.SignalAndWait có thể được sử dụng để hẹn gặp một cặp chuỗi, Wait and Pulse cũng vậy. Trong ví dụ sau, có thể nói rằng chúng tôi mô phỏng hai ManualResetEvents (nói cách khác, chúng tôi xác định hai cờ boolean!) Và sau đó thực hiện tín hiệu và chờ đối ứng bằng cách đặt một cờ trong khi chờ đợi cờ kia. Trong trường hợp này, chúng ta không cần tính nguyên tử thực sự trong tín hiệu và chờ đợi, vì vậy có thể tránh nhu cầu về “cam kết hai pha”. Miễn là chúng tôi đặt cờ đúng và Chờ trong cùng một câu lệnh khóa, điểm hẹn sẽ hoạt động:

class Rendezvous {
static object locker = new object ();
tín hiệu bool tĩnh1, signal2;

static void Main () {
// Đưa mỗi luồng vào chế độ ngủ một khoảng thời gian ngẫu nhiên.
Random r = new Random ();
new Thread (Mate) .Start (r.Next (10000));
Thread.Sleep (r.Next (10000));

khóa (khóa) {
signal1 = true;
Monitor.Pulse (tủ khóa);
while (! signal2) Monitor.Wait (khóa);
}
Console.Write (“Mate!”);
}

// Điều này được gọi thông qua một
void Mate tĩnh ParameterizedThreadStart (đối tượng delay) {
Thread.Sleep ((int) trì hoãn);
khóa (khóa) {
signal2 = true;
Monitor.Pulse (tủ khóa);
while (! signal1) Monitor.Wait (khóa);
}
Console.Write (“Mate!”);
}
}

Wait and Pulse so với Wait Handles

Bởi vì Wait và Pulse là cấu trúc đồng bộ hóa linh hoạt nhất, chúng có thể được sử dụng trong hầu hết mọi tình huống. Tay cầm Chờ, tuy nhiên, có hai ưu điểm:

Ngoài ra, Wait Handles có khả năng tương tác cao hơn theo nghĩa là chúng có thể được chuyển qua các đối số phương thức. Trong tích hợp luồng, kỹ thuật này được sử dụng hữu ích.

Về mặt hiệu suất, Wait and Pulse có một chút lợi thế, nếu tuân theo mô hình thiết kế được đề xuất cho chờ đợi, đó là:

lock (khóa)
while (điều kiện khóa) Monitor.Wait (khóa);

và điều kiện chặn xảy ra với false ngay từ đầu. Chi phí duy nhất sau đó phát sinh là việc lấy ra khóa (hàng chục nano giây) so với vài micro giây mà nó sẽ mất để gọi WaitHandle.WaitOne. Tất nhiên, điều này giả định rằng khóa là không theo ý muốn; ngay cả sự tranh cãi về khóa ngắn nhất cũng là quá đủ để giải quyết mọi chuyện; tranh chấp về khóa thường xuyên sẽ giúp Xử lý Chờ nhanh hơn!

Một hướng dẫn hợp lý là sử dụng Wait Handle trong đó một cấu trúc cụ thể tự nhiên phù hợp với công việc, nếu không, hãy sử dụng Wait and Pulse.

(Sưu tầm)