1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
|
/* __ *\
** ________ ___ / / ___ Scala API **
** / __/ __// _ | / / / _ | (c) 2003-2009, LAMP/EPFL **
** __\ \/ /__/ __ |/ /__/ __ | http://scala-lang.org/ **
** /____/\___/_/ |_/____/_/ | | **
** |/ **
\* */
// $Id$
package scala.concurrent
/** This class ...
*
* @author Martin Odersky
* @version 1.0, 12/03/2003
*/
//class MailBox with Monitor with LinkedListQueueCreator {
class MailBox extends AnyRef with ListQueueCreator {
type Message = AnyRef
private abstract class PreReceiver {
var msg: Message = null
def isDefinedAt(msg: Message): Boolean
}
private class Receiver[A](receiver: PartialFunction[Message, A]) extends PreReceiver {
def isDefinedAt(msg: Message) = receiver.isDefinedAt(msg)
def receive(): A = synchronized {
while (msg eq null) wait()
receiver(msg)
}
def receiveWithin(msec: Long): A = synchronized {
if (msg eq null) wait(msec)
receiver(if (msg ne null) msg else TIMEOUT)
}
}
private val messageQueue = queueCreate[Message]
private val receiverQueue = queueCreate[PreReceiver]
/** Unconsumed messages. */
private var sent = messageQueue.make
/** Pending receivers. */
private var receivers = receiverQueue.make
/**
* Check whether the receiver can be applied to an unconsumed message.
* If yes, the message is extracted and associated with the receiver.
* Otherwise the receiver is appended to the list of pending receivers.
*/
private def scanSentMsgs[A](receiver: Receiver[A]): Unit = synchronized {
messageQueue.extractFirst(sent, msg => receiver.isDefinedAt(msg)) match {
case None =>
receivers = receiverQueue.append(receivers, receiver)
case Some((msg, withoutMsg)) =>
sent = withoutMsg
receiver.msg = msg
}
}
/**
* First check whether a pending receiver is applicable to the sent
* message. If yes, the receiver is notified. Otherwise the message
* is appended to the linked list of sent messages.
*/
def send(msg: Message): Unit = synchronized {
receiverQueue.extractFirst(receivers, r => r.isDefinedAt(msg)) match {
case None =>
sent = messageQueue.append(sent, msg)
case Some((receiver, withoutReceiver)) =>
receivers = withoutReceiver
receiver.msg = msg
receiver synchronized { receiver.notify() }
}
}
/**
* Block until there is a message in the mailbox for which the processor
* <code>f</code> is defined.
*/
def receive[A](f: PartialFunction[Message, A]): A = {
val r = new Receiver(f)
scanSentMsgs(r)
r.receive()
}
/**
* Block until there is a message in the mailbox for which the processor
* <code>f</code> is defined or the timeout is over.
*/
def receiveWithin[A](msec: Long)(f: PartialFunction[Message, A]): A = {
val r = new Receiver(f)
scanSentMsgs(r)
r.receiveWithin(msec)
}
}
/////////////////////////////////////////////////////////////////
/**
* Module for dealing with queues.
*/
trait QueueModule[A] {
/** Type of queues. */
type T
/** Create an empty queue. */
def make: T
/** Append an element to a queue. */
def append(l: T, x: A): T
/** Extract an element satisfying a predicate from a queue. */
def extractFirst(l: T, p: A => Boolean): Option[(A, T)]
}
/** Inefficient but simple queue module creator. */
trait ListQueueCreator {
def queueCreate[A]: QueueModule[A] = new QueueModule[A] {
type T = List[A]
def make: T = Nil
def append(l: T, x: A): T = l ::: x :: Nil
def extractFirst(l: T, p: A => Boolean): Option[(A, T)] =
l match {
case Nil => None
case head :: tail =>
if (p(head))
Some((head, tail))
else
extractFirst(tail, p) match {
case None => None
case Some((x, without_x)) => Some((x, head :: without_x))
}
}
}
}
/** Efficient queue module creator based on linked lists. */
trait LinkedListQueueCreator {
import scala.collection.mutable.LinkedList
def queueCreate[A >: Null <: AnyRef]: QueueModule[A] = new QueueModule[A] {
type T = (LinkedList[A], LinkedList[A]) // fst = the list, snd = last elem
def make: T = {
val l = new LinkedList[A](null, null)
(l, l)
}
def append(l: T, x: A): T = {
val atTail = new LinkedList(x, null)
l._2 append atTail;
(l._1, atTail)
}
def extractFirst(l: T, p: A => Boolean): Option[(A, T)] = {
var xs = l._1
var xs1 = xs.next
while ((xs1 ne null) && !p(xs1.elem)) {
xs = xs1
xs1 = xs1.next
}
if (xs1 ne null) {
xs.next = xs1.next
if (xs.next eq null)
Some((xs1.elem, (l._1, xs)))
else
Some((xs1.elem, l))
}
else
None
}
}
}
|
