Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
nfs-proxy: introduce nfsv4.1 based proxy
it will re-use existing mover if found. This allows on writes to continue on the same mover if client due-to network errors fall back to IO through MDS. The implementation uses only single slot per on the server. As a result all IO requests for a single client will be serialized. The NFS door update to provide layouts for NFSv4.0 clients as well. The modified CLOSE operation is required to return layout on close for v4.0 clients. This is a first dirty implementation, but good enough to replace DCAP based proxy. Target: master Acked-by: Paul Millar Require-notes: yes Require-book: no
- Loading branch information
Showing
7 changed files
with
487 additions
and
40 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
248 changes: 248 additions & 0 deletions
248
modules/dcache-nfs/src/main/java/org/dcache/chimera/nfsv41/door/proxy/NfsProxyIo.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,248 @@ | ||
| package org.dcache.chimera.nfsv41.door.proxy; | ||
|
|
||
| import com.google.common.util.concurrent.ThreadFactoryBuilder; | ||
| import java.io.IOException; | ||
| import java.net.InetSocketAddress; | ||
| import java.nio.ByteBuffer; | ||
| import java.util.UUID; | ||
| import java.util.concurrent.Executors; | ||
| import java.util.concurrent.ScheduledExecutorService; | ||
| import java.util.concurrent.TimeUnit; | ||
|
|
||
| import org.dcache.util.NetworkUtils; | ||
|
|
||
| import org.dcache.nfs.nfsstat; | ||
| import org.dcache.nfs.v4.client.CompoundBuilder; | ||
| import org.dcache.nfs.v4.xdr.COMPOUND4args; | ||
| import org.dcache.nfs.v4.xdr.COMPOUND4res; | ||
| import org.dcache.nfs.v4.xdr.clientid4; | ||
| import org.dcache.nfs.v4.xdr.nfs_fh4; | ||
| import org.dcache.nfs.v4.xdr.nfs4_prot; | ||
| import org.dcache.nfs.v4.xdr.nfs_opnum4; | ||
| import org.dcache.nfs.v4.xdr.nfs_resop4; | ||
| import org.dcache.nfs.v4.xdr.sequenceid4; | ||
| import org.dcache.nfs.v4.xdr.sessionid4; | ||
| import org.dcache.nfs.v4.xdr.stateid4; | ||
| import org.dcache.nfs.v4.xdr.state_protect_how4; | ||
| import org.dcache.nfs.vfs.Inode; | ||
| import org.dcache.xdr.IpProtocolType; | ||
| import org.dcache.xdr.OncRpcException; | ||
| import org.dcache.xdr.OncRpcClient; | ||
| import org.dcache.xdr.RpcAuth; | ||
| import org.dcache.xdr.RpcAuthTypeUnix; | ||
| import org.dcache.xdr.RpcCall; | ||
| import org.dcache.xdr.XdrTransport; | ||
|
|
||
| /** | ||
| * A {@link ProxyIoAdapter} which proxies requests to an other NFSv4.1 server. | ||
| */ | ||
| public class NfsProxyIo implements ProxyIoAdapter { | ||
|
|
||
| // FIXME: for now we will use only a single slot. e.q serialize all requests | ||
| private final static int SLOT_ID = 0; | ||
| private final static int MAX_SLOT_ID = 0; | ||
|
|
||
| private final static int ROOT_UID = 0; | ||
| private final static int ROOT_GID = 0; | ||
| private final static int[] ROOT_GIDS = new int[0]; | ||
|
|
||
| private final static String IMPL_DOMAIN = "dCache.ORG"; | ||
| private final static String IMPL_NAME = "proxyio-nfs-client"; | ||
|
|
||
| /** | ||
| * Most up-to-date seqid for a given stateid as defined by rfc5661. | ||
| */ | ||
| private final static int SEQ_UP_TO_DATE = 0; | ||
|
|
||
| private clientid4 _clientIdByServer; | ||
| private sequenceid4 _sequenceID; | ||
| private sessionid4 _sessionid; | ||
|
|
||
| private final stateid4 stateid; | ||
| private final nfs_fh4 fh; | ||
|
|
||
| private final InetSocketAddress remoteClient; | ||
| private final RpcCall client; | ||
| private final OncRpcClient rpcClient; | ||
| private final XdrTransport transport; | ||
| private final ScheduledExecutorService sessionThread; | ||
|
|
||
| public NfsProxyIo(InetSocketAddress poolAddress, InetSocketAddress remoteClient, Inode inode, stateid4 stateid, long size) throws IOException { | ||
| this.remoteClient = remoteClient; | ||
| rpcClient = new OncRpcClient(poolAddress, IpProtocolType.TCP); | ||
| transport = rpcClient.connect(); | ||
|
|
||
| RpcAuth credential = new RpcAuthTypeUnix(ROOT_UID, ROOT_GID, ROOT_GIDS, | ||
| (int) (System.currentTimeMillis() / 1000), | ||
| NetworkUtils.getCanonicalHostName()); | ||
| client = new RpcCall(nfs4_prot.NFS4_PROGRAM, nfs4_prot.NFS_V4, credential, transport); | ||
| sessionThread = Executors.newSingleThreadScheduledExecutor( | ||
| new ThreadFactoryBuilder() | ||
| .setNameFormat("proxy-nfs-session-" + poolAddress.getAddress().getHostAddress() + "-%d") | ||
| .build() | ||
| ); | ||
|
|
||
| exchange_id(); | ||
| create_session(); | ||
| fh = new nfs_fh4(inode.toNfsHandle()); | ||
| this.stateid = new stateid4(stateid.other, SEQ_UP_TO_DATE); | ||
| } | ||
|
|
||
| @Override | ||
| public synchronized int read(ByteBuffer dst, long position) throws IOException { | ||
|
|
||
| int needToRead = dst.remaining(); | ||
| COMPOUND4args args = new CompoundBuilder() | ||
| .withSequence(false, _sessionid, _sequenceID.value, SLOT_ID, MAX_SLOT_ID) | ||
| .withPutfh(fh) | ||
| .withRead(dst.remaining(), position, stateid) | ||
| .withTag("pNFS read") | ||
| .build(); | ||
| COMPOUND4res compound4res = sendCompound(args); | ||
| dst.put(compound4res.resarray.get(2).opread.resok4.data); | ||
| return needToRead - dst.remaining(); | ||
| } | ||
|
|
||
| @Override | ||
| public synchronized int write(ByteBuffer src, long position) throws IOException { | ||
|
|
||
| byte[] data = new byte[src.remaining()]; | ||
| src.get(data); | ||
|
|
||
| COMPOUND4args args = new CompoundBuilder() | ||
| .withSequence(false, _sessionid, _sequenceID.value, SLOT_ID, MAX_SLOT_ID) | ||
| .withPutfh(fh) | ||
| .withWrite(position, data, stateid) | ||
| .withTag("pNFS write") | ||
| .build(); | ||
|
|
||
| COMPOUND4res compound4res = sendCompound(args); | ||
| return compound4res.resarray.get(2).opwrite.resok4.count.value; | ||
| } | ||
|
|
||
| @Override | ||
| public String toString() { | ||
| return String.format(" OS=%s, cl=[%s], pool=[%s]", | ||
| stateid, | ||
| remoteClient.getAddress().getHostAddress(), | ||
| transport.getRemoteSocketAddress().getAddress().getHostAddress()); | ||
| } | ||
|
|
||
| @Override | ||
| public long size() { | ||
| return 0L; | ||
| } | ||
|
|
||
| @Override | ||
| public int getSessionId() { | ||
| // forced by interface | ||
| throw new UnsupportedOperationException("Not supported yet."); | ||
| } | ||
|
|
||
| @Override | ||
| public void close() throws IOException { | ||
| sessionThread.shutdown(); | ||
| try { | ||
| destroy_session(); | ||
| } finally { | ||
| rpcClient.close(); | ||
| } | ||
| } | ||
| /** | ||
| * Call remote procedure nfsProcCompound. | ||
| * | ||
| * @param arg parameter (of type COMPOUND4args) to the remote procedure | ||
| * call. | ||
| * @return Result from remote procedure call (of type COMPOUND4res). | ||
| * @throws OncRpcException if an ONC/RPC error occurs. | ||
| * @throws IOException if an I/O error occurs. | ||
| */ | ||
| public COMPOUND4res nfsProcCompound(COMPOUND4args arg) | ||
| throws OncRpcException, IOException { | ||
| COMPOUND4res result = new COMPOUND4res(); | ||
|
|
||
| client.call(nfs4_prot.NFSPROC4_COMPOUND_4, arg, result); | ||
|
|
||
| return result; | ||
| } | ||
|
|
||
| public XdrTransport getTransport() { | ||
| return client.getTransport(); | ||
| } | ||
|
|
||
| private COMPOUND4res sendCompound(COMPOUND4args compound4args) | ||
| throws OncRpcException, IOException { | ||
|
|
||
| COMPOUND4res compound4res = nfsProcCompound(compound4args); | ||
| processSequence(compound4res); | ||
| nfsstat.throwIfNeeded(compound4res.status); | ||
| return compound4res; | ||
| } | ||
|
|
||
| private synchronized void exchange_id() throws OncRpcException, IOException { | ||
|
|
||
| COMPOUND4args args = new CompoundBuilder() | ||
| .withExchangeId(IMPL_DOMAIN, IMPL_NAME, UUID.randomUUID().toString(), 0, state_protect_how4.SP4_NONE) | ||
| .withTag("exchange_id") | ||
| .build(); | ||
|
|
||
| COMPOUND4res compound4res = sendCompound(args); | ||
|
|
||
| _clientIdByServer = compound4res.resarray.get(0).opexchange_id.eir_resok4.eir_clientid; | ||
| _sequenceID = compound4res.resarray.get(0).opexchange_id.eir_resok4.eir_sequenceid; | ||
|
|
||
| if ((compound4res.resarray.get(0).opexchange_id.eir_resok4.eir_flags.value | ||
| & nfs4_prot.EXCHGID4_FLAG_USE_PNFS_DS) == 0) { | ||
| throw new IOException("remote server is not a DS"); | ||
| } | ||
| } | ||
|
|
||
| private synchronized void create_session() throws OncRpcException, IOException { | ||
|
|
||
| COMPOUND4args args = new CompoundBuilder() | ||
| .withCreatesession(_clientIdByServer, _sequenceID) | ||
| .withTag("create_session") | ||
| .build(); | ||
|
|
||
| COMPOUND4res compound4res = sendCompound(args); | ||
|
|
||
| _sessionid = compound4res.resarray.get(0).opcreate_session.csr_resok4.csr_sessionid; | ||
| _sequenceID.value = 0; | ||
|
|
||
| sessionThread.scheduleAtFixedRate(() -> { | ||
| try { | ||
| this.sequence(); | ||
| } catch (IOException ex) { | ||
| // | ||
| } | ||
| }, | ||
| 60, 60, TimeUnit.SECONDS); | ||
| } | ||
|
|
||
| public void processSequence(COMPOUND4res compound4res) { | ||
|
|
||
| nfs_resop4 res = compound4res.resarray.get(0); | ||
| if (res.resop == nfs_opnum4.OP_SEQUENCE && res.opsequence.sr_status == nfsstat.NFS_OK) { | ||
| ++_sequenceID.value; | ||
| } | ||
| } | ||
|
|
||
| private synchronized void sequence() throws OncRpcException, IOException { | ||
|
|
||
| COMPOUND4args args = new CompoundBuilder() | ||
| .withSequence(false, _sessionid, _sequenceID.value, SLOT_ID, MAX_SLOT_ID) | ||
| .withTag("sequence") | ||
| .build(); | ||
| COMPOUND4res compound4res = sendCompound(args); | ||
| } | ||
|
|
||
| private synchronized void destroy_session() throws OncRpcException, IOException { | ||
|
|
||
| COMPOUND4args args = new CompoundBuilder() | ||
| .withDestroysession(_sessionid) | ||
| .withTag("destroy_session") | ||
| .build(); | ||
|
|
||
| COMPOUND4res compound4res = sendCompound(args); | ||
| } | ||
| } |
Oops, something went wrong.