forked from kangjianwei/LearningJDK
-
Notifications
You must be signed in to change notification settings - Fork 0
/
Cancellable.java
154 lines (132 loc) · 5.5 KB
/
Cancellable.java
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
/*
* Copyright (c) 2008, 2009, Oracle and/or its affiliates. All rights reserved.
* DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
*
* This code is free software; you can redistribute it and/or modify it
* under the terms of the GNU General Public License version 2 only, as
* published by the Free Software Foundation. Oracle designates this
* particular file as subject to the "Classpath" exception as provided
* by Oracle in the LICENSE file that accompanied this code.
*
* This code is distributed in the hope that it will be useful, but WITHOUT
* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
* FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License
* version 2 for more details (a copy is included in the LICENSE file that
* accompanied this code).
*
* You should have received a copy of the GNU General Public License version
* 2 along with this work; if not, write to the Free Software Foundation,
* Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA.
*
* Please contact Oracle, 500 Oracle Parkway, Redwood Shores, CA 94065 USA
* or visit www.oracle.com if you need additional information or have any
* questions.
*/
package sun.nio.fs;
import jdk.internal.misc.Unsafe;
import java.util.concurrent.ExecutionException;
/**
* Base implementation of a task (typically native) that polls a memory location during execution so that it may be aborted/cancelled before completion.
* The task is executed by invoking the {@link runInterruptibly} method defined here and cancelled by invoking Thread.interrupt.
*/
// 可取消的任务,该任务受线程中断标记的影响
abstract class Cancellable implements Runnable {
private static final Unsafe unsafe = Unsafe.getUnsafe();
private final Object lock = new Object();
private final long pollingAddress; // 指向本地内存,保存"取消"状态码
// the following require lock when examining or changing
private boolean completed; // 当前任务是否已完成
private Throwable exception; // 异常信息
protected Cancellable() {
// 申请4字节的本地内存,并返回分配的内存地址
pollingAddress = unsafe.allocateMemory(4);
// 置空
unsafe.putIntVolatile(null, pollingAddress, 0);
}
@Override
public final void run() {
try {
// 执行当前任务
implRun();
} catch(Throwable t) {
synchronized(lock) {
exception = t;
}
} finally {
synchronized(lock) {
completed = true;
unsafe.freeMemory(pollingAddress);
}
}
}
/**
* The task body. This should periodically poll the memory location to check for cancellation.
*/
// 对当前任务执行的具体实现由不同的平台自行实现
abstract void implRun() throws Throwable;
/**
* Returns the memory address of a 4-byte int that should be polled to detect cancellation.
*/
protected long addressToPollForCancel() {
return pollingAddress;
}
/**
* The value to write to the polled memory location to indicate that the task has been cancelled.
* If this method is not overridden then it defaults to MAX_VALUE.
*/
// 返回取消任务时的状态码,可覆盖此实现
protected int cancelValue() {
return Integer.MAX_VALUE;
}
/**
* Invokes the given task in its own thread. If this (meaning the current)
* thread is interrupted then an attempt is make to cancel the background
* thread by writing into the memory location that it polls cooperatively.
*/
// 执行指定的任务,会阻塞发起当前操作的线程;该任务会响应线程中断
static void runInterruptibly(Cancellable task) throws ExecutionException {
// 启动任务
Thread copyThread = new Thread(null, task, "NIO-Task", 0, false);
copyThread.start();
boolean cancelledByInterrupt = false;
while(copyThread.isAlive()) {
try {
// 使copyThread线程进入WAITING状态,直到copyThread线程执行完成,或被中断之后,再唤醒runInterruptibly()的调用者所在的线程
copyThread.join();
} catch(InterruptedException e) {
cancelledByInterrupt = true;
task.cancel(); // 响应中断,停止copyThread
}
}
if(cancelledByInterrupt) {
// 中断线程(只是给线程预设一个标记,不是立即让线程停下来)
Thread.currentThread().interrupt();
}
// 如果发生异常,需要抛出
Throwable exc = task.exception();
if(exc != null) {
throw new ExecutionException(exc);
}
}
/**
* "cancels" the task by writing bits into memory location that it polled by the task.
*/
// 取消当前任务
final void cancel() {
synchronized(lock) {
if(!completed) {
// 保存"取消"状态码
unsafe.putIntVolatile(null, pollingAddress, cancelValue());
}
}
}
/**
* Returns the exception thrown by the task or null if the task completed successfully.
*/
// 返回执行过程中的异常信息
private Throwable exception() {
synchronized(lock) {
return exception;
}
}
}