author | wenzelm |
Tue, 07 Sep 2021 15:15:13 +0200 | |
changeset 74254 | 53e1a14e2ef1 |
parent 74253 | 45dc9de1bd33 |
child 75393 | 87ebf5a50283 |
permissions | -rw-r--r-- |
56695
963732291084
consumer thread with unbounded queueing of requests (similar to Message_Channel in ML);
wenzelm
parents:
diff
changeset
|
1 |
/* Title: Pure/Concurrent/consumer_thread.scala |
963732291084
consumer thread with unbounded queueing of requests (similar to Message_Channel in ML);
wenzelm
parents:
diff
changeset
|
2 |
Author: Makarius |
963732291084
consumer thread with unbounded queueing of requests (similar to Message_Channel in ML);
wenzelm
parents:
diff
changeset
|
3 |
|
56702
f96ad2b19c38
support for requests with explicit acknowledgment (and exception propagation);
wenzelm
parents:
56701
diff
changeset
|
4 |
Consumer thread with unbounded queueing of requests, and optional |
f96ad2b19c38
support for requests with explicit acknowledgment (and exception propagation);
wenzelm
parents:
56701
diff
changeset
|
5 |
acknowledgment. |
56695
963732291084
consumer thread with unbounded queueing of requests (similar to Message_Channel in ML);
wenzelm
parents:
diff
changeset
|
6 |
*/ |
963732291084
consumer thread with unbounded queueing of requests (similar to Message_Channel in ML);
wenzelm
parents:
diff
changeset
|
7 |
|
963732291084
consumer thread with unbounded queueing of requests (similar to Message_Channel in ML);
wenzelm
parents:
diff
changeset
|
8 |
package isabelle |
963732291084
consumer thread with unbounded queueing of requests (similar to Message_Channel in ML);
wenzelm
parents:
diff
changeset
|
9 |
|
963732291084
consumer thread with unbounded queueing of requests (similar to Message_Channel in ML);
wenzelm
parents:
diff
changeset
|
10 |
|
963732291084
consumer thread with unbounded queueing of requests (similar to Message_Channel in ML);
wenzelm
parents:
diff
changeset
|
11 |
import scala.annotation.tailrec |
963732291084
consumer thread with unbounded queueing of requests (similar to Message_Channel in ML);
wenzelm
parents:
diff
changeset
|
12 |
|
963732291084
consumer thread with unbounded queueing of requests (similar to Message_Channel in ML);
wenzelm
parents:
diff
changeset
|
13 |
|
56696 | 14 |
object Consumer_Thread |
56695
963732291084
consumer thread with unbounded queueing of requests (similar to Message_Channel in ML);
wenzelm
parents:
diff
changeset
|
15 |
{ |
71143
5ea3ed3c52b3
support for bulk operations: consume mailbox content in batches;
wenzelm
parents:
71142
diff
changeset
|
16 |
def fork_bulk[A](name: String = "", daemon: Boolean = false)( |
5ea3ed3c52b3
support for bulk operations: consume mailbox content in batches;
wenzelm
parents:
71142
diff
changeset
|
17 |
bulk: A => Boolean, |
5ea3ed3c52b3
support for bulk operations: consume mailbox content in batches;
wenzelm
parents:
71142
diff
changeset
|
18 |
consume: List[A] => (List[Exn.Result[Unit]], Boolean), |
5ea3ed3c52b3
support for bulk operations: consume mailbox content in batches;
wenzelm
parents:
71142
diff
changeset
|
19 |
finish: () => Unit = () => ()): Consumer_Thread[A] = |
5ea3ed3c52b3
support for bulk operations: consume mailbox content in batches;
wenzelm
parents:
71142
diff
changeset
|
20 |
new Consumer_Thread[A](name, daemon, bulk, consume, finish) |
5ea3ed3c52b3
support for bulk operations: consume mailbox content in batches;
wenzelm
parents:
71142
diff
changeset
|
21 |
|
56698 | 22 |
def fork[A](name: String = "", daemon: Boolean = false)( |
23 |
consume: A => Boolean, |
|
24 |
finish: () => Unit = () => ()): Consumer_Thread[A] = |
|
71143
5ea3ed3c52b3
support for bulk operations: consume mailbox content in batches;
wenzelm
parents:
71142
diff
changeset
|
25 |
{ |
5ea3ed3c52b3
support for bulk operations: consume mailbox content in batches;
wenzelm
parents:
71142
diff
changeset
|
26 |
def consume_single(args: List[A]): (List[Exn.Result[Unit]], Boolean) = |
5ea3ed3c52b3
support for bulk operations: consume mailbox content in batches;
wenzelm
parents:
71142
diff
changeset
|
27 |
{ |
5ea3ed3c52b3
support for bulk operations: consume mailbox content in batches;
wenzelm
parents:
71142
diff
changeset
|
28 |
assert(args.length == 1) |
5ea3ed3c52b3
support for bulk operations: consume mailbox content in batches;
wenzelm
parents:
71142
diff
changeset
|
29 |
Exn.capture { consume(args.head) } match { |
5ea3ed3c52b3
support for bulk operations: consume mailbox content in batches;
wenzelm
parents:
71142
diff
changeset
|
30 |
case Exn.Res(continue) => (List(Exn.Res(())), continue) |
5ea3ed3c52b3
support for bulk operations: consume mailbox content in batches;
wenzelm
parents:
71142
diff
changeset
|
31 |
case Exn.Exn(exn) => (List(Exn.Exn(exn)), true) |
5ea3ed3c52b3
support for bulk operations: consume mailbox content in batches;
wenzelm
parents:
71142
diff
changeset
|
32 |
} |
5ea3ed3c52b3
support for bulk operations: consume mailbox content in batches;
wenzelm
parents:
71142
diff
changeset
|
33 |
} |
5ea3ed3c52b3
support for bulk operations: consume mailbox content in batches;
wenzelm
parents:
71142
diff
changeset
|
34 |
|
5ea3ed3c52b3
support for bulk operations: consume mailbox content in batches;
wenzelm
parents:
71142
diff
changeset
|
35 |
fork_bulk(name = name, daemon = daemon)(_ => false, consume_single, finish = finish) |
5ea3ed3c52b3
support for bulk operations: consume mailbox content in batches;
wenzelm
parents:
71142
diff
changeset
|
36 |
} |
56696 | 37 |
} |
56695
963732291084
consumer thread with unbounded queueing of requests (similar to Message_Channel in ML);
wenzelm
parents:
diff
changeset
|
38 |
|
56698 | 39 |
final class Consumer_Thread[A] private( |
71143
5ea3ed3c52b3
support for bulk operations: consume mailbox content in batches;
wenzelm
parents:
71142
diff
changeset
|
40 |
name: String, daemon: Boolean, |
5ea3ed3c52b3
support for bulk operations: consume mailbox content in batches;
wenzelm
parents:
71142
diff
changeset
|
41 |
bulk: A => Boolean, |
5ea3ed3c52b3
support for bulk operations: consume mailbox content in batches;
wenzelm
parents:
71142
diff
changeset
|
42 |
consume: List[A] => (List[Exn.Result[Unit]], Boolean), |
5ea3ed3c52b3
support for bulk operations: consume mailbox content in batches;
wenzelm
parents:
71142
diff
changeset
|
43 |
finish: () => Unit) |
56696 | 44 |
{ |
71142 | 45 |
/* thread */ |
46 |
||
56698 | 47 |
private var active = true |
71142 | 48 |
private val mailbox = Mailbox[Option[Request]] |
56696 | 49 |
|
71692 | 50 |
private val thread = Isabelle_Thread.fork(name = name, daemon = daemon) { main_loop(Nil) } |
74253 | 51 |
def is_active(): Boolean = active && thread.isAlive |
74254 | 52 |
def check_thread(): Boolean = Thread.currentThread == thread |
56708 | 53 |
|
56701 | 54 |
private def failure(exn: Throwable): Unit = |
56782
433cf57550fa
more systematic Isabelle output, like in classic Isabelle/ML (without markup);
wenzelm
parents:
56718
diff
changeset
|
55 |
Output.error_message( |
56708 | 56 |
"Consumer thread failure: " + quote(thread.getName) + "\n" + Exn.message(exn)) |
56701 | 57 |
|
58 |
private def robust_finish(): Unit = |
|
59 |
try { finish() } catch { case exn: Throwable => failure(exn) } |
|
60 |
||
71142 | 61 |
|
62 |
/* requests */ |
|
63 |
||
71143
5ea3ed3c52b3
support for bulk operations: consume mailbox content in batches;
wenzelm
parents:
71142
diff
changeset
|
64 |
private class Request(val arg: A, acknowledge: Boolean = false) |
71142 | 65 |
{ |
71143
5ea3ed3c52b3
support for bulk operations: consume mailbox content in batches;
wenzelm
parents:
71142
diff
changeset
|
66 |
val ack: Option[Synchronized[Option[Exn.Result[Unit]]]] = |
71142 | 67 |
if (acknowledge) Some(Synchronized(None)) else None |
68 |
||
74252 | 69 |
def await(): Unit = |
71142 | 70 |
{ |
71 |
for (a <- ack) { |
|
72 |
Exn.release(a.guarded_access({ case None => None case res => Some((res.get, res)) })) |
|
73 |
} |
|
74 |
} |
|
75 |
} |
|
76 |
||
73340 | 77 |
private def request(req: Request): Unit = |
71142 | 78 |
{ |
79 |
synchronized { |
|
74253 | 80 |
if (is_active()) mailbox.send(Some(req)) |
71142 | 81 |
else error("Consumer thread not active: " + quote(thread.getName)) |
82 |
} |
|
74252 | 83 |
req.await() |
71142 | 84 |
} |
85 |
||
86 |
@tailrec private def main_loop(msgs: List[Option[Request]]): Unit = |
|
57417
29fe9bac501b
more tight Mailbox: single list is sufficient for single receiver, reverse outside critical section;
wenzelm
parents:
56782
diff
changeset
|
87 |
msgs match { |
71144 | 88 |
case Nil => main_loop(mailbox.receive()) |
57417
29fe9bac501b
more tight Mailbox: single list is sufficient for single receiver, reverse outside critical section;
wenzelm
parents:
56782
diff
changeset
|
89 |
case None :: _ => robust_finish() |
71143
5ea3ed3c52b3
support for bulk operations: consume mailbox content in batches;
wenzelm
parents:
71142
diff
changeset
|
90 |
case _ => |
5ea3ed3c52b3
support for bulk operations: consume mailbox content in batches;
wenzelm
parents:
71142
diff
changeset
|
91 |
val reqs = |
5ea3ed3c52b3
support for bulk operations: consume mailbox content in batches;
wenzelm
parents:
71142
diff
changeset
|
92 |
proper_list(msgs.takeWhile(msg => msg.isDefined && bulk(msg.get.arg))) |
5ea3ed3c52b3
support for bulk operations: consume mailbox content in batches;
wenzelm
parents:
71142
diff
changeset
|
93 |
.getOrElse(msgs.take(1)) |
5ea3ed3c52b3
support for bulk operations: consume mailbox content in batches;
wenzelm
parents:
71142
diff
changeset
|
94 |
.map(_.get) |
5ea3ed3c52b3
support for bulk operations: consume mailbox content in batches;
wenzelm
parents:
71142
diff
changeset
|
95 |
|
5ea3ed3c52b3
support for bulk operations: consume mailbox content in batches;
wenzelm
parents:
71142
diff
changeset
|
96 |
val (results, continue) = consume(reqs.map(_.arg)) |
5ea3ed3c52b3
support for bulk operations: consume mailbox content in batches;
wenzelm
parents:
71142
diff
changeset
|
97 |
|
5ea3ed3c52b3
support for bulk operations: consume mailbox content in batches;
wenzelm
parents:
71142
diff
changeset
|
98 |
for { (Some(req), Some(res)) <- reqs.map(Some(_)).zipAll(results.map(Some(_)), None, None) } |
5ea3ed3c52b3
support for bulk operations: consume mailbox content in batches;
wenzelm
parents:
71142
diff
changeset
|
99 |
{ |
5ea3ed3c52b3
support for bulk operations: consume mailbox content in batches;
wenzelm
parents:
71142
diff
changeset
|
100 |
(req.ack, res) match { |
74251 | 101 |
case (Some(a), _) => a.change(_ => Some(res)) |
102 |
case (None, Exn.Res(_)) => |
|
103 |
case (None, Exn.Exn(exn)) => failure(exn) |
|
71143
5ea3ed3c52b3
support for bulk operations: consume mailbox content in batches;
wenzelm
parents:
71142
diff
changeset
|
104 |
} |
5ea3ed3c52b3
support for bulk operations: consume mailbox content in batches;
wenzelm
parents:
71142
diff
changeset
|
105 |
} |
5ea3ed3c52b3
support for bulk operations: consume mailbox content in batches;
wenzelm
parents:
71142
diff
changeset
|
106 |
|
71231
dafa5fce70f1
clarified messages streaming (again, amending 5ea3ed3c52b3): avoid too many small messages stacking up, e.g. when loading HOL-Analysis.Analysis.thy into Isabelle/jEdit;
wenzelm
parents:
71144
diff
changeset
|
107 |
if (continue) main_loop(msgs.drop(reqs.length)) |
71143
5ea3ed3c52b3
support for bulk operations: consume mailbox content in batches;
wenzelm
parents:
71142
diff
changeset
|
108 |
else robust_finish() |
56695
963732291084
consumer thread with unbounded queueing of requests (similar to Message_Channel in ML);
wenzelm
parents:
diff
changeset
|
109 |
} |
963732291084
consumer thread with unbounded queueing of requests (similar to Message_Channel in ML);
wenzelm
parents:
diff
changeset
|
110 |
|
56701 | 111 |
|
112 |
/* main methods */ |
|
113 |
||
74253 | 114 |
assert(is_active()) |
56696 | 115 |
|
73340 | 116 |
def send(arg: A): Unit = request(new Request(arg)) |
117 |
def send_wait(arg: A): Unit = request(new Request(arg, acknowledge = true)) |
|
56702
f96ad2b19c38
support for requests with explicit acknowledgment (and exception propagation);
wenzelm
parents:
56701
diff
changeset
|
118 |
|
73340 | 119 |
def shutdown(): Unit = |
56696 | 120 |
{ |
74253 | 121 |
synchronized { if (is_active()) { active = false; mailbox.send(None) } } |
74140 | 122 |
thread.join() |
56696 | 123 |
} |
56695
963732291084
consumer thread with unbounded queueing of requests (similar to Message_Channel in ML);
wenzelm
parents:
diff
changeset
|
124 |
} |