Skip to content

Commit 43abbc9

Browse files
committed
Add Windows keyed event implementation of ThreadParker
1 parent cf64fd9 commit 43abbc9

File tree

5 files changed

+321
-83
lines changed

5 files changed

+321
-83
lines changed

Cargo.toml

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -15,6 +15,10 @@ smallvec = "0.1"
1515
[target.'cfg(target_os = "linux")'.dependencies]
1616
libc = "0.2"
1717

18+
[target.'cfg(windows)'.dependencies]
19+
winapi = "0.2"
20+
kernel32-sys = "0.2"
21+
1822
[dev-dependencies]
1923
rand = "0.3"
2024
lazy_static = "0.2"

src/lib.rs

Lines changed: 9 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -111,13 +111,21 @@ extern crate lazy_static;
111111
#[cfg(all(feature = "nightly", target_os = "linux"))]
112112
extern crate libc;
113113

114+
#[cfg(windows)]
115+
extern crate winapi;
116+
#[cfg(windows)]
117+
extern crate kernel32;
118+
114119
// Spin limit from JikesRVM & Webkit experiments
115120
const SPIN_LIMIT: usize = 40;
116121

117122
#[cfg(all(feature = "nightly", target_os = "linux"))]
118123
#[path = "thread_parker/linux.rs"]
119124
mod thread_parker;
120-
#[cfg(not(all(feature = "nightly", target_os = "linux")))]
125+
#[cfg(windows)]
126+
#[path = "thread_parker/windows.rs"]
127+
mod thread_parker;
128+
#[cfg(not(any(windows, all(feature = "nightly", target_os = "linux"))))]
121129
#[path = "thread_parker/generic.rs"]
122130
mod thread_parker;
123131

src/parking_lot.rs

Lines changed: 63 additions & 63 deletions
Original file line numberDiff line numberDiff line change
@@ -268,81 +268,81 @@ pub unsafe fn park(key: usize,
268268
timeout: Option<Instant>)
269269
-> bool {
270270
// Grab our thread data, this also ensures that the hash table exists
271-
THREAD_DATA.with(|thread_data| {
272-
// Lock the bucket for the given key
273-
let bucket = lock_bucket(key).unwrap();
271+
let thread_data = &*THREAD_DATA.with(|x| x as *const ThreadData);
274272

275-
// If the validation function fails, just return
276-
if !validate() {
277-
bucket.mutex.unlock();
278-
return false;
279-
}
273+
// Lock the bucket for the given key
274+
let bucket = lock_bucket(key).unwrap();
280275

281-
// Append our thread data to the queue and unlock the bucket
282-
thread_data.next_in_queue.set(ptr::null());
283-
thread_data.key.set(key);
284-
thread_data.parker.prepare_park();
285-
if !bucket.queue_head.get().is_null() {
286-
(*bucket.queue_tail.get()).next_in_queue.set(thread_data);
287-
} else {
288-
bucket.queue_head.set(thread_data);
289-
}
290-
bucket.queue_tail.set(thread_data);
276+
// If the validation function fails, just return
277+
if !validate() {
291278
bucket.mutex.unlock();
279+
return false;
280+
}
292281

293-
// Invoke the pre-sleep callback
294-
before_sleep();
295-
296-
// Park our thread and determine whether we were woken up by an unpark
297-
// or by our timeout. Note that this isn't precise: we can still be
298-
// unparked since we are still in the queue.
299-
let unparked = match timeout {
300-
Some(timeout) => thread_data.parker.park_until(timeout),
301-
None => {
302-
thread_data.parker.park();
303-
true
304-
}
305-
};
282+
// Append our thread data to the queue and unlock the bucket
283+
thread_data.next_in_queue.set(ptr::null());
284+
thread_data.key.set(key);
285+
thread_data.parker.prepare_park();
286+
if !bucket.queue_head.get().is_null() {
287+
(*bucket.queue_tail.get()).next_in_queue.set(thread_data);
288+
} else {
289+
bucket.queue_head.set(thread_data);
290+
}
291+
bucket.queue_tail.set(thread_data);
292+
bucket.mutex.unlock();
306293

307-
// If we were unparked, return now
308-
if unparked {
309-
return true;
294+
// Invoke the pre-sleep callback
295+
before_sleep();
296+
297+
// Park our thread and determine whether we were woken up by an unpark or by
298+
// our timeout. Note that this isn't precise: we can still be unparked
299+
// since we are still in the queue.
300+
let unparked = match timeout {
301+
Some(timeout) => thread_data.parker.park_until(timeout),
302+
None => {
303+
thread_data.parker.park();
304+
true
310305
}
306+
};
311307

312-
// Lock our bucket again. Note that the hashtable may have been rehashed
313-
// in the meantime.
314-
let bucket = lock_bucket(key).unwrap();
308+
// If we were unparked, return now
309+
if unparked {
310+
return true;
311+
}
315312

316-
// Now we need to check again if we were unparked or timed out. Unlike
317-
// the last check this is precise because we hold the bucket lock.
318-
if !thread_data.parker.timed_out() {
319-
bucket.mutex.unlock();
320-
return true;
321-
}
313+
// Lock our bucket again. Note that the hashtable may have been rehashed in
314+
// the meantime.
315+
let bucket = lock_bucket(key).unwrap();
322316

323-
// We timed out, so we now need to remove our thread from the queue
324-
let mut link = &bucket.queue_head;
325-
let mut current = bucket.queue_head.get();
326-
let mut previous = ptr::null();
327-
while !current.is_null() {
328-
if current == thread_data {
329-
let next = (*current).next_in_queue.get();
330-
link.set(next);
331-
if bucket.queue_tail.get() == current {
332-
bucket.queue_tail.set(previous);
333-
}
334-
break;
335-
} else {
336-
link = &(*current).next_in_queue;
337-
previous = current;
338-
current = link.get();
317+
// Now we need to check again if we were unparked or timed out. Unlike the
318+
// last check this is precise because we hold the bucket lock.
319+
if !thread_data.parker.timed_out() {
320+
bucket.mutex.unlock();
321+
return true;
322+
}
323+
324+
// We timed out, so we now need to remove our thread from the queue
325+
let mut link = &bucket.queue_head;
326+
let mut current = bucket.queue_head.get();
327+
let mut previous = ptr::null();
328+
while !current.is_null() {
329+
if current == thread_data {
330+
let next = (*current).next_in_queue.get();
331+
link.set(next);
332+
if bucket.queue_tail.get() == current {
333+
bucket.queue_tail.set(previous);
339334
}
335+
break;
336+
} else {
337+
link = &(*current).next_in_queue;
338+
previous = current;
339+
current = link.get();
340340
}
341+
}
341342

342-
// Unlock the bucket, we are done
343-
bucket.mutex.unlock();
344-
false
345-
})
343+
// Unlock the bucket, we are done
344+
bucket.mutex.unlock();
345+
false
346346
}
347347

348348
/// Unparks one thread from the queue associated with the given key.

0 commit comments

Comments
 (0)