forked from Alluxio/alluxio
-
Notifications
You must be signed in to change notification settings - Fork 0
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
- Loading branch information
1 parent
ceeeefe
commit 1472081
Showing
10 changed files
with
748 additions
and
0 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
95 changes: 95 additions & 0 deletions
95
dora/core/client/fs/src/main/java/alluxio/client/file/dora/ucx/UcxDataReader.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,95 @@ | ||
| package alluxio.client.file.dora.ucx; | ||
|
|
||
| import alluxio.PositionReader; | ||
| import alluxio.file.ByteBufferTargetBuffer; | ||
| import alluxio.file.ReadTargetBuffer; | ||
| import alluxio.proto.dataserver.Protocol; | ||
| import alluxio.wire.WorkerNetAddress; | ||
| import alluxio.worker.ucx.UcpProxy; | ||
|
|
||
| import com.google.common.base.Preconditions; | ||
| import org.openucx.jucx.UcxCallback; | ||
| import org.openucx.jucx.UcxException; | ||
| import org.openucx.jucx.UcxUtils; | ||
| import org.openucx.jucx.ucp.UcpEndpoint; | ||
| import org.openucx.jucx.ucp.UcpEndpointParams; | ||
| import org.openucx.jucx.ucp.UcpMemory; | ||
| import org.openucx.jucx.ucp.UcpRequest; | ||
| import org.openucx.jucx.ucp.UcpWorker; | ||
| import org.slf4j.Logger; | ||
| import org.slf4j.LoggerFactory; | ||
|
|
||
| import java.io.IOException; | ||
| import java.net.InetSocketAddress; | ||
| import java.nio.ByteBuffer; | ||
| import java.util.function.Supplier; | ||
|
|
||
| public class UcxDataReader implements PositionReader { | ||
| private static final Logger LOG = LoggerFactory.getLogger(UcxDataReader.class); | ||
|
|
||
| InetSocketAddress mAddr; | ||
| UcpWorker mWorker; | ||
| UcpEndpoint mWorkerEndpoint; | ||
| Supplier<Protocol.ReadRequest.Builder> mRequestBuilder; | ||
| public UcxDataReader(InetSocketAddress addr, UcpWorker worker, | ||
| Protocol.ReadRequest.Builder requestBuilder) { | ||
| mAddr = addr; | ||
| mWorker = worker; | ||
| mRequestBuilder = requestBuilder::clone; | ||
| } | ||
|
|
||
| public void acquireServerConn() { | ||
| if (mWorkerEndpoint != null) { | ||
| return; | ||
| } | ||
| mWorkerEndpoint = mWorker.newEndpoint( | ||
| new UcpEndpointParams() | ||
| .setPeerErrorHandlingMode() | ||
| .setErrorHandler((ep, status, errorMsg) -> | ||
| System.out.println("[ERROR] creating ep to remote:" | ||
| + mAddr + " errored out: " + errorMsg | ||
| + " status:" + status + ",ep:" + ep.toString())) | ||
| .setSocketAddress(mAddr)); | ||
| } | ||
|
|
||
| public void waitForRequest(UcpRequest ucpRequest) { | ||
| while(!ucpRequest.isCompleted()) { | ||
| try { | ||
| mWorker.progress(); | ||
| } catch (Exception e) { | ||
| LOG.error("Error progressing req:", e); | ||
| } | ||
| } | ||
| } | ||
|
|
||
| @Override | ||
| public int readInternal(long position, ReadTargetBuffer buffer, int length) throws IOException { | ||
| Protocol.ReadRequest.Builder builder = mRequestBuilder.get() | ||
| .setLength(length) | ||
| .setOffset(position) | ||
| .clearCancel(); | ||
| Protocol.ReadRequest readRequest = builder.build(); | ||
| byte[] serializedBytes = readRequest.toByteArray(); | ||
| ByteBuffer buf = ByteBuffer.allocateDirect(serializedBytes.length); | ||
| buf.put(serializedBytes); | ||
| buf.rewind(); | ||
| UcpRequest sendRequest = mWorkerEndpoint.sendTaggedNonBlocking(buf, serializedBytes.length, new UcxCallback() { | ||
| public void onSuccess(UcpRequest request) { | ||
| LOG.info("ReadReq:{} sent.", readRequest); | ||
| } | ||
|
|
||
| public void onError(int ucsStatus, String errorMsg) { | ||
| throw new UcxException(errorMsg); | ||
| } | ||
| }); | ||
| waitForRequest(sendRequest); | ||
| // now wait to recv data | ||
| Preconditions.checkArgument((buffer instanceof ByteBufferTargetBuffer | ||
| && buffer.byteBuffer().isDirect()), | ||
| "Must be ByteBufferTargetBuffer with direct ByteBuffer"); | ||
| UcpRequest recvRequest = mWorker.recvTaggedNonBlocking( | ||
| UcxUtils.getAddress(buffer.byteBuffer()), length, 0,0, null); | ||
| waitForRequest(recvRequest); | ||
| return 0; | ||
| } | ||
| } |
40 changes: 40 additions & 0 deletions
40
dora/core/client/fs/src/main/java/alluxio/worker/ucx/UcpProxy.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,40 @@ | ||
| package alluxio.worker.ucx; | ||
|
|
||
| import org.openucx.jucx.ucp.UcpContext; | ||
| import org.openucx.jucx.ucp.UcpParams; | ||
| import org.openucx.jucx.ucp.UcpWorker; | ||
| import org.openucx.jucx.ucp.UcpWorkerParams; | ||
|
|
||
| import java.util.concurrent.locks.ReentrantLock; | ||
|
|
||
| public class UcpProxy { | ||
|
|
||
| private static final UcpContext sGlobalContext = new UcpContext(new UcpParams() | ||
| .requestStreamFeature() | ||
| .requestTagFeature() | ||
| .requestWakeupFeature()); | ||
| public UcpWorker mWorker; | ||
| public static UcpProxy sUcpProxy; | ||
| private static ReentrantLock sInstanceLock = new ReentrantLock(); | ||
|
|
||
| public static UcpProxy getInstance() { | ||
| if (sUcpProxy != null) { | ||
| return sUcpProxy; | ||
| } | ||
| sInstanceLock.lock(); | ||
| try { | ||
| if (sUcpProxy != null) { | ||
| return sUcpProxy; | ||
| } | ||
| sUcpProxy = new UcpProxy(); | ||
| return sUcpProxy; | ||
| } finally { | ||
| sInstanceLock.unlock(); | ||
| } | ||
| } | ||
|
|
||
| public UcpProxy() { | ||
| mWorker = sGlobalContext.newWorker(new UcpWorkerParams()); | ||
| } | ||
|
|
||
| } |
Oops, something went wrong.