big-data/hadoop
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

HADOOP-6609. Fixed deadlock in RPC by replacing shared static

Browse Source 
DataOutputBuffer in the UTF8 class with a thread local variable. (omalley)


git-svn-id: https://svn.apache.org/repos/asf/hadoop/common/trunk@918880 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
Owen O'Malley 2010-03-04 07:20:29 +00:00
parent 393baff732
commit b1ec35e60c
2 changed files with 37 additions and 30 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
CHANGES.txt
View File

@ -185,8 +185,8 @@ Trunk (unreleased changes)
HADOOP-6599 Split existing RpcMetrics into RpcMetrics & RpcDetailedMetrics.
(Suresh Srinivas via Sanjay Radia)
HADOOP-6537 Declare more detailed exceptions in FileContext and AbstractFileSystem
(Suresh Srinivas via Sanjay Radia)
HADOOP-6537 Declare more detailed exceptions in FileContext and
AbstractFileSystem (Suresh Srinivas via Sanjay Radia)
OPTIMIZATIONS
@ -264,6 +264,9 @@ Trunk (unreleased changes)
HADOOP-6545. Changes the Key for the FileSystem cache to be UGI (ddas)
HADOOP-6609. Fixed deadlock in RPC by replacing shared static
DataOutputBuffer in the UTF8 class with a thread local variable. (omalley)
Release 0.21.0 - Unreleased
INCOMPATIBLE CHANGES

60
src/java/org/apache/hadoop/io/UTF8.java
View File

@ -33,9 +33,16 @@
*/
public class UTF8 implements WritableComparable {
private static final Log LOG= LogFactory.getLog(UTF8.class);
private static final DataOutputBuffer OBUF = new DataOutputBuffer();
private static final DataInputBuffer IBUF = new DataInputBuffer();
private static final ThreadLocal<DataOutputBuffer> OBUF_FACTORY =
new ThreadLocal<DataOutputBuffer>(){
@Override
protected DataOutputBuffer initialValue() {
return new DataOutputBuffer();
}
};
private static final byte[] EMPTY_BYTES = new byte[0];
private byte[] bytes = EMPTY_BYTES;
@ -81,11 +88,10 @@ public void set(String string) {
bytes = new byte[length];
try { // avoid sync'd allocations
synchronized (OBUF) {
OBUF.reset();
writeChars(OBUF, string, 0, string.length());
System.arraycopy(OBUF.getData(), 0, bytes, 0, length);
}
DataOutputBuffer obuf = OBUF_FACTORY.get();
obuf.reset();
writeChars(obuf, string, 0, string.length());
System.arraycopy(obuf.getData(), 0, bytes, 0, length);
} catch (IOException e) {
throw new RuntimeException(e);
}
@ -182,11 +188,10 @@ public int compare(byte[] b1, int s1, int l1,
public static byte[] getBytes(String string) {
byte[] result = new byte[utf8Length(string)];
try { // avoid sync'd allocations
synchronized (OBUF) {
OBUF.reset();
writeChars(OBUF, string, 0, string.length());
System.arraycopy(OBUF.getData(), 0, result, 0, OBUF.getLength());
}
DataOutputBuffer obuf = OBUF_FACTORY.get();
obuf.reset();
writeChars(obuf, string, 0, string.length());
System.arraycopy(obuf.getData(), 0, result, 0, obuf.getLength());
} catch (IOException e) {
throw new RuntimeException(e);
}
@ -206,23 +211,22 @@ public static String readString(DataInput in) throws IOException {
private static void readChars(DataInput in, StringBuffer buffer, int nBytes)
throws IOException {
synchronized (OBUF) {
OBUF.reset();
OBUF.write(in, nBytes);
byte[] bytes = OBUF.getData();
int i = 0;
while (i < nBytes) {
byte b = bytes[i++];
if ((b & 0x80) == 0) {
buffer.append((char)(b & 0x7F));
} else if ((b & 0xE0) != 0xE0) {
buffer.append((char)(((b & 0x1F) << 6)
| (bytes[i++] & 0x3F)));
} else {
buffer.append((char)(((b & 0x0F) << 12)
| ((bytes[i++] & 0x3F) << 6)
| (bytes[i++] & 0x3F)));
}
DataOutputBuffer obuf = OBUF_FACTORY.get();
obuf.reset();
obuf.write(in, nBytes);
byte[] bytes = obuf.getData();
int i = 0;
while (i < nBytes) {
byte b = bytes[i++];
if ((b & 0x80) == 0) {
buffer.append((char)(b & 0x7F));
} else if ((b & 0xE0) != 0xE0) {
buffer.append((char)(((b & 0x1F) << 6)
| (bytes[i++] & 0x3F)));
} else {
buffer.append((char)(((b & 0x0F) << 12)
| ((bytes[i++] & 0x3F) << 6)
| (bytes[i++] & 0x3F)));
}
}
}