Skip to content

Commit 73fed01

Browse files
committed
Add task::worker. Spawns a task and returns a channel to it
It takes a lot of boilerplate to create a task and establish a way to talk to it. This function simplifies that, allowing you to write something like 'worker(f).chan <| start'. Implementation is very unsafe and only works for a few types of channels, but something like this is very useful.
1 parent b82bedb commit 73fed01

File tree

2 files changed

+93
-0
lines changed

2 files changed

+93
-0
lines changed

src/lib/task.rs

Lines changed: 70 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -52,6 +52,76 @@ fn clone_chan[T](chan[T] c) -> chan[T] {
5252
ret unsafe::reinterpret_cast(cloned);
5353
}
5454

55+
// Spawn a task and immediately return a channel for communicating to it
56+
fn worker[T](fn(port[T]) f) -> rec(task task, chan[T] chan) {
57+
// FIXME: This is frighteningly unsafe and only works for
58+
// a few cases
59+
60+
type opaque = int;
61+
62+
// FIXME: This terrible hackery is because worktask can't currently
63+
// have type params
64+
type wordsz1 = int;
65+
type wordsz2 = rec(int a, int b);
66+
type wordsz3 = rec(int a, int b, int c);
67+
type wordsz4 = rec(int a, int b, int c, int d);
68+
type opaquechan_1wordsz = chan[chan[wordsz1]];
69+
type opaquechan_2wordsz = chan[chan[wordsz2]];
70+
type opaquechan_3wordsz = chan[chan[wordsz3]];
71+
type opaquechan_4wordsz = chan[chan[wordsz4]];
72+
73+
fn worktask1(opaquechan_1wordsz setupch, opaque fptr) {
74+
let *fn(port[wordsz1]) f = unsafe::reinterpret_cast(fptr);
75+
auto p = port[wordsz1]();
76+
setupch <| chan(p);
77+
(*f)(p);
78+
}
79+
80+
fn worktask2(opaquechan_2wordsz setupch, opaque fptr) {
81+
let *fn(port[wordsz2]) f = unsafe::reinterpret_cast(fptr);
82+
auto p = port[wordsz2]();
83+
setupch <| chan(p);
84+
(*f)(p);
85+
}
86+
87+
fn worktask3(opaquechan_3wordsz setupch, opaque fptr) {
88+
let *fn(port[wordsz3]) f = unsafe::reinterpret_cast(fptr);
89+
auto p = port[wordsz3]();
90+
setupch <| chan(p);
91+
(*f)(p);
92+
}
93+
94+
fn worktask4(opaquechan_4wordsz setupch, opaque fptr) {
95+
let *fn(port[wordsz4]) f = unsafe::reinterpret_cast(fptr);
96+
auto p = port[wordsz4]();
97+
setupch <| chan(p);
98+
(*f)(p);
99+
}
100+
101+
auto p = port[chan[T]]();
102+
auto setupch = chan(p);
103+
auto fptr = unsafe::reinterpret_cast(ptr::addr_of(f));
104+
105+
auto Tsz = sys::size_of[T]();
106+
auto t = if Tsz == sys::size_of[wordsz1]() {
107+
auto setupchptr = unsafe::reinterpret_cast(setupch);
108+
spawn worktask1(setupchptr, fptr)
109+
} else if Tsz == sys::size_of[wordsz2]() {
110+
auto setupchptr = unsafe::reinterpret_cast(setupch);
111+
spawn worktask2(setupchptr, fptr)
112+
} else if Tsz == sys::size_of[wordsz3]() {
113+
auto setupchptr = unsafe::reinterpret_cast(setupch);
114+
spawn worktask3(setupchptr, fptr)
115+
} else if Tsz == sys::size_of[wordsz4]() {
116+
auto setupchptr = unsafe::reinterpret_cast(setupch);
117+
spawn worktask4(setupchptr, fptr)
118+
} else {
119+
fail #fmt("unhandled type size %u in task::worker", Tsz)
120+
};
121+
auto ch; p |> ch;
122+
ret rec(task = t, chan = ch);
123+
}
124+
55125
// Local Variables:
56126
// mode: rust;
57127
// fill-column: 78;

src/test/stdtest/task.rs

Lines changed: 23 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -32,3 +32,26 @@ fn test_join() {
3232

3333
assert task::join(failtask) == task::tr_failure;
3434
}
35+
36+
#[test]
37+
fn test_worker() {
38+
task::worker(fn(port[int] p) {
39+
auto x; p |> x;
40+
assert x == 10;
41+
}).chan <| 10;
42+
43+
task::worker(fn(port[rec(int x, int y)] p) {
44+
auto x; p |> x;
45+
assert x.y == 20;
46+
}).chan <| rec(x = 10, y = 20);
47+
48+
task::worker(fn(port[rec(int x, int y, int z)] p) {
49+
auto x; p |> x;
50+
assert x.z == 30;
51+
}).chan <| rec(x = 10, y = 20, z = 30);
52+
53+
task::worker(fn(port[rec(int a, int b, int c, int d)] p) {
54+
auto x; p |> x;
55+
assert x.d == 40;
56+
}).chan <| rec(a = 10, b = 20, c = 30, d = 40);
57+
}

0 commit comments

Comments
 (0)