Skip to content
体验新版
项目
组织
正在加载...
登录
切换导航
打开侧边栏
int
Rust
提交
e633249b
R
Rust
项目概览
int
/
Rust
大约 1 年 前同步成功
通知
1
Star
0
Fork
0
代码
文件
提交
分支
Tags
贡献者
分支图
Diff
Issue
0
列表
看板
标记
里程碑
合并请求
0
DevOps
流水线
流水线任务
计划
Wiki
0
Wiki
分析
仓库
DevOps
项目成员
Pages
R
Rust
项目概览
项目概览
详情
发布
仓库
仓库
文件
提交
分支
标签
贡献者
分支图
比较
Issue
0
Issue
0
列表
看板
标记
里程碑
合并请求
0
合并请求
0
Pages
DevOps
DevOps
流水线
流水线任务
计划
分析
分析
仓库分析
DevOps
Wiki
0
Wiki
成员
成员
收起侧边栏
关闭侧边栏
动态
分支图
创建新Issue
流水线任务
提交
Issue看板
体验新版 GitCode,发现更多精彩内容 >>
提交
e633249b
编写于
2月 11, 2014
作者:
A
Alex Crichton
浏览文件
操作
浏览文件
下载
电子邮件补丁
差异文件
Test fixes and rebase conflicts
上级
0a6b9219
变更
5
隐藏空白更改
内联
并排
Showing
5 changed file
with
31 addition
and
44 deletion
+31
-44
src/libstd/comm/mod.rs
src/libstd/comm/mod.rs
+13
-13
src/libstd/comm/oneshot.rs
src/libstd/comm/oneshot.rs
+14
-27
src/libstd/io/net/tcp.rs
src/libstd/io/net/tcp.rs
+1
-1
src/libstd/io/net/udp.rs
src/libstd/io/net/udp.rs
+2
-2
src/libstd/io/net/unix.rs
src/libstd/io/net/unix.rs
+1
-1
未找到文件。
src/libstd/comm/mod.rs
浏览文件 @
e633249b
...
...
@@ -242,13 +242,13 @@
use
iter
::
Iterator
;
use
kinds
::
Send
;
use
kinds
::
marker
;
use
mem
;
use
ops
::
Drop
;
use
option
::{
Some
,
None
,
Option
};
use
result
::{
Ok
,
Err
,
Result
};
use
rt
::
local
::
Local
;
use
rt
::
task
::{
Task
,
BlockedTask
};
use
sync
::
arc
::
UnsafeArc
;
use
util
;
pub
use
comm
::
select
::{
Select
,
Handle
};
...
...
@@ -427,7 +427,7 @@ pub fn try_send(&self, t: T) -> bool {
unsafe
{
let
mut
tmp
=
Chan
::
my_new
(
Stream
(
new_inner
));
util
::
swap
(
&
mut
cast
::
transmute_mut
(
self
)
.inner
,
&
mut
tmp
.inner
);
mem
::
swap
(
&
mut
cast
::
transmute_mut
(
self
)
.inner
,
&
mut
tmp
.inner
);
}
return
ret
;
}
...
...
@@ -460,7 +460,7 @@ fn clone(&self) -> Chan<T> {
(
*
packet
.get
())
.inherit_blocker
(
sleeper
);
let
mut
tmp
=
Chan
::
my_new
(
Shared
(
packet
.clone
()));
util
::
swap
(
&
mut
cast
::
transmute_mut
(
self
)
.inner
,
&
mut
tmp
.inner
);
mem
::
swap
(
&
mut
cast
::
transmute_mut
(
self
)
.inner
,
&
mut
tmp
.inner
);
}
Chan
::
my_new
(
Shared
(
packet
))
}
...
...
@@ -556,8 +556,8 @@ pub fn try_recv(&self) -> TryRecvResult<T> {
}
};
unsafe
{
util
::
swap
(
&
mut
cast
::
transmute_mut
(
self
)
.inner
,
&
mut
new_port
.inner
);
mem
::
swap
(
&
mut
cast
::
transmute_mut
(
self
)
.inner
,
&
mut
new_port
.inner
);
}
}
}
...
...
@@ -602,8 +602,8 @@ pub fn recv_opt(&self) -> Option<T> {
}
};
unsafe
{
util
::
swap
(
&
mut
cast
::
transmute_mut
(
self
)
.inner
,
&
mut
new_port
.inner
);
mem
::
swap
(
&
mut
cast
::
transmute_mut
(
self
)
.inner
,
&
mut
new_port
.inner
);
}
}
}
...
...
@@ -636,8 +636,8 @@ fn can_recv(&self) -> bool {
}
};
unsafe
{
util
::
swap
(
&
mut
cast
::
transmute_mut
(
self
)
.inner
,
&
mut
new_port
.inner
);
mem
::
swap
(
&
mut
cast
::
transmute_mut
(
self
)
.inner
,
&
mut
new_port
.inner
);
}
}
}
...
...
@@ -665,8 +665,8 @@ fn start_selection(&self, mut task: BlockedTask) -> Result<(), BlockedTask>{
};
task
=
t
;
unsafe
{
util
::
swap
(
&
mut
cast
::
transmute_mut
(
self
)
.inner
,
&
mut
new_port
.inner
);
mem
::
swap
(
&
mut
cast
::
transmute_mut
(
self
)
.inner
,
&
mut
new_port
.inner
);
}
}
}
...
...
@@ -686,8 +686,8 @@ fn abort_selection(&self) -> bool {
let
mut
new_port
=
match
result
{
Ok
(
b
)
=>
return
b
,
Err
(
p
)
=>
p
};
was_upgrade
=
true
;
unsafe
{
util
::
swap
(
&
mut
cast
::
transmute_mut
(
self
)
.inner
,
&
mut
new_port
.inner
);
mem
::
swap
(
&
mut
cast
::
transmute_mut
(
self
)
.inner
,
&
mut
new_port
.inner
);
}
}
}
...
...
src/libstd/comm/oneshot.rs
浏览文件 @
e633249b
...
...
@@ -34,13 +34,13 @@
use
comm
::
Port
;
use
kinds
::
Send
;
use
mem
;
use
ops
::
Drop
;
use
option
::{
Some
,
None
,
Option
};
use
result
::{
Result
,
Ok
,
Err
};
use
rt
::
local
::
Local
;
use
rt
::
task
::{
Task
,
BlockedTask
};
use
sync
::
atomics
;
use
util
;
// Various states you can find a port in.
static
EMPTY
:
uint
=
0
;
...
...
@@ -100,10 +100,7 @@ pub fn send(&mut self, t: T) -> bool {
self
.data
=
Some
(
t
);
self
.upgrade
=
SendUsed
;
// This atomic swap uses a "Release" memory ordering to ensure that all
// our previous memory writes are visible to the other thread (notably
// the write of data/upgrade)
match
self
.state
.swap
(
DATA
,
atomics
::
Release
)
{
match
self
.state
.swap
(
DATA
,
atomics
::
SeqCst
)
{
// Sent the data, no one was waiting
EMPTY
=>
true
,
...
...
@@ -141,14 +138,11 @@ pub fn sent(&self) -> bool {
pub
fn
recv
(
&
mut
self
)
->
Result
<
T
,
Failure
<
T
>>
{
// Attempt to not block the task (it's a little expensive). If it looks
// like we're not empty, then immediately go through to `try_recv`.
//
// These atomics use an Acquire memory ordering in order to have all the
// previous writes of the releasing thread visible to us.
if
self
.state
.load
(
atomics
::
Acquire
)
==
EMPTY
{
if
self
.state
.load
(
atomics
::
SeqCst
)
==
EMPTY
{
let
t
:
~
Task
=
Local
::
take
();
t
.deschedule
(
1
,
|
task
|
{
let
n
=
unsafe
{
task
.cast_to_uint
()
};
match
self
.state
.compare_and_swap
(
EMPTY
,
n
,
atomics
::
Acquire
)
{
match
self
.state
.compare_and_swap
(
EMPTY
,
n
,
atomics
::
SeqCst
)
{
// Nothing on the channel, we legitimately block
EMPTY
=>
Ok
(()),
...
...
@@ -168,8 +162,7 @@ pub fn recv(&mut self) -> Result<T, Failure<T>> {
}
pub
fn
try_recv
(
&
mut
self
)
->
Result
<
T
,
Failure
<
T
>>
{
// see above for why Acquire is used.
match
self
.state
.load
(
atomics
::
Acquire
)
{
match
self
.state
.load
(
atomics
::
SeqCst
)
{
EMPTY
=>
Err
(
Empty
),
// We saw some data on the channel, but the channel can be used
...
...
@@ -179,7 +172,7 @@ pub fn try_recv(&mut self) -> Result<T, Failure<T>> {
// the state changes under our feet we'd rather just see that state
// change.
DATA
=>
{
self
.state
.compare_and_swap
(
DATA
,
EMPTY
,
atomics
::
Acquire
);
self
.state
.compare_and_swap
(
DATA
,
EMPTY
,
atomics
::
SeqCst
);
match
self
.data
.take
()
{
Some
(
data
)
=>
Ok
(
data
),
None
=>
unreachable!
(),
...
...
@@ -194,7 +187,7 @@ pub fn try_recv(&mut self) -> Result<T, Failure<T>> {
match
self
.data
.take
()
{
Some
(
data
)
=>
Ok
(
data
),
None
=>
{
match
util
::
replace
(
&
mut
self
.upgrade
,
SendUsed
)
{
match
mem
::
replace
(
&
mut
self
.upgrade
,
SendUsed
)
{
SendUsed
|
NothingSent
=>
Err
(
Disconnected
),
GoUp
(
upgrade
)
=>
Err
(
Upgraded
(
upgrade
))
}
...
...
@@ -216,9 +209,7 @@ pub fn upgrade(&mut self, up: Port<T>) -> UpgradeResult {
};
self
.upgrade
=
GoUp
(
up
);
// Use a Release memory ordering in order to make sure that our write to
// `upgrade` is visible to the other thread.
match
self
.state
.swap
(
DISCONNECTED
,
atomics
::
Release
)
{
match
self
.state
.swap
(
DISCONNECTED
,
atomics
::
SeqCst
)
{
// If the channel is empty or has data on it, then we're good to go.
// Senders will check the data before the upgrade (in case we
// plastered over the DATA state).
...
...
@@ -246,9 +237,7 @@ pub fn drop_chan(&mut self) {
}
pub
fn
drop_port
(
&
mut
self
)
{
// Use an Acquire memory ordering in order to see the data that the
// senders are sending.
match
self
.state
.swap
(
DISCONNECTED
,
atomics
::
Acquire
)
{
match
self
.state
.swap
(
DISCONNECTED
,
atomics
::
SeqCst
)
{
// An empty channel has nothing to do, and a remotely disconnected
// channel also has nothing to do b/c we're about to run the drop
// glue
...
...
@@ -271,13 +260,12 @@ pub fn drop_port(&mut self) {
// If Ok, the value is whether this port has data, if Err, then the upgraded
// port needs to be checked instead of this one.
pub
fn
can_recv
(
&
mut
self
)
->
Result
<
bool
,
Port
<
T
>>
{
// Use Acquire so we can see all previous memory writes
match
self
.state
.load
(
atomics
::
Acquire
)
{
match
self
.state
.load
(
atomics
::
SeqCst
)
{
EMPTY
=>
Ok
(
false
),
// Welp, we tried
DATA
=>
Ok
(
true
),
// we have some un-acquired data
DISCONNECTED
if
self
.data
.is_some
()
=>
Ok
(
true
),
// we have data
DISCONNECTED
=>
{
match
util
::
replace
(
&
mut
self
.upgrade
,
SendUsed
)
{
match
mem
::
replace
(
&
mut
self
.upgrade
,
SendUsed
)
{
// The other end sent us an upgrade, so we need to
// propagate upwards whether the upgrade can receive
// data
...
...
@@ -304,7 +292,7 @@ pub fn start_selection(&mut self, task: BlockedTask) -> SelectionResult<T> {
SelCanceled
(
unsafe
{
BlockedTask
::
cast_from_uint
(
n
)
})
}
DISCONNECTED
=>
{
match
util
::
replace
(
&
mut
self
.upgrade
,
SendUsed
)
{
match
mem
::
replace
(
&
mut
self
.upgrade
,
SendUsed
)
{
// The other end sent us an upgrade, so we need to
// propagate upwards whether the upgrade can receive
// data
...
...
@@ -331,8 +319,7 @@ pub fn start_selection(&mut self, task: BlockedTask) -> SelectionResult<T> {
//
// The return value indicates whether there's data on this port.
pub
fn
abort_selection
(
&
mut
self
)
->
Result
<
bool
,
Port
<
T
>>
{
// use Acquire to make sure we see all previous memory writes
let
state
=
match
self
.state
.load
(
atomics
::
Acquire
)
{
let
state
=
match
self
.state
.load
(
atomics
::
SeqCst
)
{
// Each of these states means that no further activity will happen
// with regard to abortion selection
s
@
EMPTY
|
...
...
@@ -357,7 +344,7 @@ pub fn abort_selection(&mut self) -> Result<bool, Port<T>> {
// aborted.
DISCONNECTED
=>
{
assert
!
(
self
.data
.is_none
());
match
util
::
replace
(
&
mut
self
.upgrade
,
SendUsed
)
{
match
mem
::
replace
(
&
mut
self
.upgrade
,
SendUsed
)
{
GoUp
(
port
)
=>
Err
(
port
),
_
=>
Ok
(
true
),
}
...
...
src/libstd/io/net/tcp.rs
浏览文件 @
e633249b
...
...
@@ -698,7 +698,7 @@ pub fn peer_name(addr: SocketAddr) {
iotest!
(
fn
tcp_clone_two_read
()
{
let
addr
=
next_test_ip6
();
let
mut
acceptor
=
TcpListener
::
bind
(
addr
)
.listen
();
let
(
p
,
c
)
=
Shared
Chan
::
new
();
let
(
p
,
c
)
=
Chan
::
new
();
let
c2
=
c
.clone
();
spawn
(
proc
()
{
...
...
src/libstd/io/net/udp.rs
浏览文件 @
e633249b
...
...
@@ -301,7 +301,7 @@ pub fn socket_name(addr: SocketAddr) {
let
addr2
=
next_test_ip4
();
let
mut
sock1
=
UdpSocket
::
bind
(
addr1
)
.unwrap
();
let
sock2
=
UdpSocket
::
bind
(
addr2
)
.unwrap
();
let
(
p
,
c
)
=
Shared
Chan
::
new
();
let
(
p
,
c
)
=
Chan
::
new
();
let
c2
=
c
.clone
();
spawn
(
proc
()
{
...
...
@@ -335,7 +335,7 @@ pub fn socket_name(addr: SocketAddr) {
let
mut
sock1
=
UdpSocket
::
bind
(
addr1
)
.unwrap
();
let
sock2
=
UdpSocket
::
bind
(
addr2
)
.unwrap
();
let
(
p
,
c
)
=
Shared
Chan
::
new
();
let
(
p
,
c
)
=
Chan
::
new
();
let
(
serv_port
,
serv_chan
)
=
Chan
::
new
();
spawn
(
proc
()
{
...
...
src/libstd/io/net/unix.rs
浏览文件 @
e633249b
...
...
@@ -270,7 +270,7 @@ fn unix_clone_smoke() {
fn
unix_clone_two_read
()
{
let
addr
=
next_test_unix
();
let
mut
acceptor
=
UnixListener
::
bind
(
&
addr
)
.listen
();
let
(
p
,
c
)
=
Shared
Chan
::
new
();
let
(
p
,
c
)
=
Chan
::
new
();
let
c2
=
c
.clone
();
spawn
(
proc
()
{
...
...
编辑
预览
Markdown
is supported
0%
请重试
或
添加新附件
.
添加附件
取消
You are about to add
0
people
to the discussion. Proceed with caution.
先完成此消息的编辑!
取消
想要评论请
注册
或
登录