author | wenzelm |
Mon, 28 Jun 2021 12:29:00 +0200 | |
changeset 73888 | 9c2dd041477b |
parent 73340 | 0ffcad1f6130 |
child 74140 | 8a5e02ef975c |
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) } |
56718
096139bcfadd
replaced manager Actor by Consumer_Thread, which is lazy to defer its start to actual Handler init time;
wenzelm
parents:
56708
diff
changeset
|
51 |
def is_active: Boolean = active && thread.isAlive |
66094 | 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 |
||
73340 | 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 { |
|
80 |
if (is_active) mailbox.send(Some(req)) |
|
81 |
else error("Consumer thread not active: " + quote(thread.getName)) |
|
82 |
} |
|
83 |
req.await |
|
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 { |
5ea3ed3c52b3
support for bulk operations: consume mailbox content in batches;
wenzelm
parents:
71142
diff
changeset
|
101 |
case ((Some(a), _)) => a.change(_ => Some(res)) |
5ea3ed3c52b3
support for bulk operations: consume mailbox content in batches;
wenzelm
parents:
71142
diff
changeset
|
102 |
case ((None, Exn.Res(_))) => |
5ea3ed3c52b3
support for bulk operations: consume mailbox content in batches;
wenzelm
parents:
71142
diff
changeset
|
103 |
case ((None, Exn.Exn(exn))) => failure(exn) |
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 |
||
71142 | 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 |
{ |
57417
29fe9bac501b
more tight Mailbox: single list is sufficient for single receiver, reverse outside critical section;
wenzelm
parents:
56782
diff
changeset
|
121 |
synchronized { if (is_active) { active = false; mailbox.send(None) } } |
56696 | 122 |
thread.join |
123 |
} |
|
56695
963732291084
consumer thread with unbounded queueing of requests (similar to Message_Channel in ML);
wenzelm
parents:
diff
changeset
|
124 |
} |