Skip to content
体验新版
项目
组织
正在加载...
登录
切换导航
打开侧边栏
xxadev
jenkins
提交
4c10efe1
J
jenkins
项目概览
xxadev
/
jenkins
与 Fork 源项目一致
从无法访问的项目Fork
通知
3
Star
0
Fork
0
代码
文件
提交
分支
Tags
贡献者
分支图
Diff
Issue
0
列表
看板
标记
里程碑
合并请求
0
Wiki
0
Wiki
分析
仓库
DevOps
项目成员
Pages
J
jenkins
项目概览
项目概览
详情
发布
仓库
仓库
文件
提交
分支
标签
贡献者
分支图
比较
Issue
0
Issue
0
列表
看板
标记
里程碑
合并请求
0
合并请求
0
Pages
分析
分析
仓库分析
DevOps
Wiki
0
Wiki
成员
成员
收起侧边栏
关闭侧边栏
动态
分支图
创建新Issue
提交
Issue看板
体验新版 GitCode,发现更多精彩内容 >>
提交
4c10efe1
编写于
2月 16, 2011
作者:
B
Brian Atkinson
浏览文件
操作
浏览文件
下载
差异文件
Merge branch 'JENKINS-8592'
上级
be5438b3
1d1c1889
变更
6
显示空白变更内容
内联
并排
Showing
6 changed file
with
119 addition
and
73 deletion
+119
-73
remoting/src/main/java/hudson/remoting/Channel.java
remoting/src/main/java/hudson/remoting/Channel.java
+5
-4
remoting/src/main/java/hudson/remoting/Command.java
remoting/src/main/java/hudson/remoting/Command.java
+8
-0
remoting/src/main/java/hudson/remoting/Pipe.java
remoting/src/main/java/hudson/remoting/Pipe.java
+1
-13
remoting/src/main/java/hudson/remoting/PipeWindow.java
remoting/src/main/java/hudson/remoting/PipeWindow.java
+34
-5
remoting/src/main/java/hudson/remoting/ProxyOutputStream.java
...ting/src/main/java/hudson/remoting/ProxyOutputStream.java
+37
-2
remoting/src/test/java/hudson/remoting/PipeTest.java
remoting/src/test/java/hudson/remoting/PipeTest.java
+34
-49
未找到文件。
remoting/src/main/java/hudson/remoting/Channel.java
浏览文件 @
4c10efe1
...
...
@@ -599,9 +599,6 @@ public class Channel implements VirtualChannel, IChannel {
}
/*package*/
PipeWindow
getPipeWindow
(
int
oid
)
{
if
(!
remoteCapability
.
supportsPipeThrottling
())
return
PipeWindow
.
FAKE
;
synchronized
(
pipeWindows
)
{
Key
k
=
new
Key
(
oid
);
WeakReference
<
PipeWindow
>
v
=
pipeWindows
.
get
(
k
);
...
...
@@ -611,7 +608,11 @@ public class Channel implements VirtualChannel, IChannel {
return
w
;
}
Real
w
=
new
Real
(
k
,
PIPE_WINDOW_SIZE
);
PipeWindow
w
;
if
(
remoteCapability
.
supportsPipeThrottling
())
w
=
new
Real
(
k
,
PIPE_WINDOW_SIZE
);
else
w
=
new
PipeWindow
.
Fake
();
pipeWindows
.
put
(
k
,
new
WeakReference
<
PipeWindow
>(
w
));
return
w
;
}
...
...
remoting/src/main/java/hudson/remoting/Command.java
浏览文件 @
4c10efe1
...
...
@@ -47,6 +47,10 @@ abstract class Command implements Serializable {
this
(
true
);
}
protected
Command
(
Throwable
cause
)
{
this
.
createdAt
=
new
Source
(
cause
);
}
/**
* @param recordCreatedAt
* If false, skip the recording of where the command is created. This makes the trouble-shooting
...
...
@@ -74,6 +78,10 @@ abstract class Command implements Serializable {
public
Source
()
{
}
private
Source
(
Throwable
cause
)
{
super
(
cause
);
}
public
String
toString
()
{
return
"Command "
+
Command
.
this
.
toString
()+
" created at"
;
}
...
...
remoting/src/main/java/hudson/remoting/Pipe.java
浏览文件 @
4c10efe1
...
...
@@ -93,19 +93,7 @@ public final class Pipe implements Serializable {
* Gets the reading end of the pipe.
*/
public
InputStream
getIn
()
{
return
new
FilterInputStream
(
in
)
{
@Override
public
void
close
()
throws
IOException
{
try
{
// Since closing the reading side does not stop the writing side. We read till the stream is done.
final
byte
[]
buffer
=
new
byte
[
4096
];
while
(
read
(
buffer
)
!=
-
1
)
{
}
}
finally
{
super
.
close
();
}
}
};
return
in
;
}
/**
...
...
remoting/src/main/java/hudson/remoting/PipeWindow.java
浏览文件 @
4c10efe1
...
...
@@ -23,6 +23,7 @@
*/
package
hudson.remoting
;
import
java.io.IOException
;
import
java.io.OutputStream
;
import
java.util.logging.Logger
;
...
...
@@ -47,21 +48,46 @@ import static java.util.logging.Level.*;
* @author Kohsuke Kawaguchi
*/
abstract
class
PipeWindow
{
protected
Throwable
dead
;
abstract
void
increase
(
int
delta
);
abstract
int
peek
();
/**
* Blocks until some space becomes available.
*
* @throws IOException
* If we learned that there is an irrecoverable problem on the remote side that prevents us from writing.
* @throws InterruptedException
* If a thread was interrupted while blocking.
*/
abstract
int
get
()
throws
InterruptedException
;
abstract
int
get
()
throws
InterruptedException
,
IOException
;
abstract
void
decrease
(
int
delta
);
/**
* Indicates that the remote end has died and all the further send attempt should fail.
*/
void
dead
(
Throwable
cause
)
{
this
.
dead
=
cause
;
}
/**
* If we already know that the remote end had developed a problem, throw an exception.
* Otherwise no-op.
*/
protected
void
checkDeath
()
throws
IOException
{
if
(
dead
!=
null
)
// the remote end failed to write.
throw
(
IOException
)
new
IOException
(
"Pipe is already closed"
).
initCause
(
dead
);
}
/**
* Fake implementation used when the receiver side doesn't support throttling.
*/
static
final
PipeWindow
FAKE
=
new
PipeWindow
()
{
static
class
Fake
extends
PipeWindow
{
void
increase
(
int
delta
)
{
}
...
...
@@ -69,13 +95,14 @@ abstract class PipeWindow {
return
Integer
.
MAX_VALUE
;
}
int
get
()
throws
InterruptedException
{
int
get
()
throws
InterruptedException
,
IOException
{
checkDeath
();
return
Integer
.
MAX_VALUE
;
}
void
decrease
(
int
delta
)
{
}
}
;
}
static
final
class
Key
{
public
final
int
oid
;
...
...
@@ -134,13 +161,15 @@ abstract class PipeWindow {
* to avoid fragmenting the window size. That is, if a bunch of small ACKs come in a sequence,
* bundle them up into a bigger size before making a call.
*/
public
int
get
()
throws
InterruptedException
{
public
int
get
()
throws
InterruptedException
,
IOException
{
checkDeath
();
synchronized
(
this
)
{
if
(
available
>
0
)
return
available
;
while
(
available
<=
0
)
{
wait
();
checkDeath
();
}
}
...
...
remoting/src/main/java/hudson/remoting/ProxyOutputStream.java
浏览文件 @
4c10efe1
...
...
@@ -184,12 +184,23 @@ final class ProxyOutputStream extends OutputStream {
try
{
os
.
write
(
buf
);
}
catch
(
IOException
e
)
{
try
{
channel
.
send
(
new
NotifyDeadWriter
(
e
,
oid
));
}
catch
(
ChannelClosedException
x
)
{
// the other direction can be already closed if the connection
// shut down is initiated from this side. In that case, remain silent.
}
catch
(
IOException
x
)
{
// ignore errors
LOGGER
.
log
(
Level
.
WARNING
,
"Failed to write to stream"
,
e
);
LOGGER
.
log
(
Level
.
WARNING
,
"Failed to notify the sender that the write end is dead"
,
x
);
LOGGER
.
log
(
Level
.
WARNING
,
"... the failed write was:"
,
e
);
}
}
finally
{
if
(
channel
.
remoteCapability
.
supportsPipeThrottling
())
{
try
{
channel
.
send
(
new
Ack
(
oid
,
buf
.
length
));
}
catch
(
ChannelClosedException
x
)
{
// the other direction can be already closed if the connection
// shut down is initiated from this side. In that case, remain silent.
}
catch
(
IOException
e
)
{
// ignore errors
LOGGER
.
log
(
Level
.
WARNING
,
"Failed to ack the stream"
,
e
);
...
...
@@ -329,5 +340,29 @@ final class ProxyOutputStream extends OutputStream {
private
static
final
long
serialVersionUID
=
1L
;
}
/**
* {@link Command} to notify the sender that the receiver is dead.
*/
private
static
final
class
NotifyDeadWriter
extends
Command
{
private
final
int
oid
;
private
NotifyDeadWriter
(
Throwable
cause
,
int
oid
)
{
super
(
cause
);
this
.
oid
=
oid
;
}
@Override
protected
void
execute
(
Channel
channel
)
{
PipeWindow
w
=
channel
.
getPipeWindow
(
oid
);
w
.
dead
(
createdAt
.
getCause
());
}
public
String
toString
()
{
return
"Pipe.Dead("
+
oid
+
")"
;
}
private
static
final
long
serialVersionUID
=
1L
;
}
private
static
final
Logger
LOGGER
=
Logger
.
getLogger
(
ProxyOutputStream
.
class
.
getName
());
}
remoting/src/test/java/hudson/remoting/PipeTest.java
浏览文件 @
4c10efe1
...
...
@@ -24,13 +24,11 @@
package
hudson.remoting
;
import
hudson.remoting.ChannelRunner.InProcessCompatibilityMode
;
import
hudson.remoting.FastPipedInputStream.ClosedBy
;
import
junit.framework.Test
;
import
org.apache.commons.io.IOUtils
;
import
org.apache.commons.io.output.NullOutputStream
;
import
org.jvnet.hudson.test.Bug
;
import
org.jvnet.hudson.test.For
;
import
org.jvnet.hudson.test.Url
;
import
java.io.ByteArrayOutputStream
;
import
java.io.DataInputStream
;
...
...
@@ -39,9 +37,7 @@ import java.io.IOException;
import
java.io.InputStream
;
import
java.io.Serializable
;
import
java.util.Arrays
;
import
java.util.logging.Handler
;
import
java.util.logging.LogRecord
;
import
java.util.logging.Logger
;
import
java.util.concurrent.ExecutionException
;
/**
* Test {@link Pipe}.
...
...
@@ -64,57 +60,46 @@ public class PipeTest extends RmiTestBase implements Serializable {
}
/**
* Helper class for testPartialReadQuietlyConsumesPipeOnClose.
* Have the reader close the read end of the pipe while the writer is still writing.
* The writer should pick up a failure.
*/
private
static
class
EventCounterHandler
extends
Handler
{
int
count
=
0
;
@Bug
(
8592
)
@For
(
Pipe
.
class
)
public
void
testReaderCloseWhileWriterIsStillWriting
()
throws
Exception
{
final
Pipe
p
=
Pipe
.
createRemoteToLocal
();
final
Future
<
Void
>
f
=
channel
.
callAsync
(
new
InfiniteWriter
(
p
));
final
InputStream
in
=
p
.
getIn
();
assertEquals
(
in
.
read
(),
0
);
in
.
close
();
@Override
public
void
publish
(
final
LogRecord
record
)
{
if
(
ProxyOutputStream
.
class
.
getName
().
equals
(
record
.
getLoggerName
()))
{
Throwable
thrown
=
record
.
getThrown
();
while
(
thrown
!=
null
)
{
if
(
thrown
instanceof
ClosedBy
)
{
count
+=
1
;
break
;
}
thrown
=
thrown
.
getCause
();
try
{
f
.
get
();
fail
();
}
catch
(
ExecutionException
e
)
{
// should have resulted in an IOException
if
(!(
e
.
getCause
()
instanceof
IOException
))
{
e
.
printStackTrace
();
fail
();
}
}
}
@Override
public
void
flush
()
{
}
/**
* Just writes forever to the pipe
*/
private
static
class
InfiniteWriter
implements
Callable
<
Void
,
Exception
>
{
private
final
Pipe
pipe
;
@Override
public
void
close
()
throws
SecurityException
{
public
InfiniteWriter
(
Pipe
pipe
)
{
this
.
pipe
=
pipe
;
}
public
int
getCount
()
{
return
count
;
public
Void
call
()
throws
Exception
{
while
(
true
)
{
pipe
.
getOut
().
write
(
0
);
Thread
.
sleep
(
10
);
}
}
/**
* This test should be reproducing the initial bug as reported in JENKINS-8592. The assert in this test is not the
* best and is fragile, but it is the best I can come up with.
*/
@Bug
(
8592
)
@For
(
Pipe
.
class
)
@Url
(
"http://issues.jenkins-ci.org/browse/JENKINS-8592"
)
public
void
testPartialReadQuietlyConsumesPipeOnClose
()
throws
Exception
{
final
EventCounterHandler
handler
=
new
EventCounterHandler
();
final
Logger
logger
=
Logger
.
getLogger
(
ProxyOutputStream
.
class
.
getName
());
logger
.
addHandler
(
handler
);
final
Pipe
p
=
Pipe
.
createRemoteToLocal
();
final
Future
<
Integer
>
f
=
channel
.
callAsync
(
new
WritingCallable
(
p
));
final
InputStream
in
=
p
.
getIn
();
assertEquals
(
in
.
read
(),
0
);
in
.
close
();
f
.
get
();
logger
.
removeHandler
(
handler
);
assertEquals
(
0
,
handler
.
getCount
());
}
private
static
class
WritingCallable
implements
Callable
<
Integer
,
IOException
>
{
...
...
编辑
预览
Markdown
is supported
0%
请重试
或
添加新附件
.
添加附件
取消
You are about to add
0
people
to the discussion. Proceed with caution.
先完成此消息的编辑!
取消
想要评论请
注册
或
登录