diff --git a/CMakeLists.txt b/CMakeLists.txt index 6b9886c7..f87b0453 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -16,18 +16,28 @@ set(INSTALL_CMAKE_DIR "${INSTALL_LIB_DIR}/cmake/libnfs" CACHE PATH "Installation option(BUILD_SHARED_LIBS "Build shared libraries" ON) option(ENABLE_TESTS "Build and run test programs" OFF) option(ENABLE_DOCUMENTATION "Build Documentation" OFF) +option(ENABLE_PARANOID "Enable paranoid checking in the code" OFF) option(ENABLE_UTILS "Build util programs" OFF) option(ENABLE_EXAMPLES "Build example programs" OFF) option(ENABLE_MULTITHREADING "Enable multithreading support" OFF) +option(ENABLE_INSECURE_AUTH_FOR_DEVTEST "Enable AZAUTH for non-TLS connections" OFF) if(ENABLE_TESTS) set(ENABLE_UTILS ON CACHE BOOL "Building utils required by tests" FORCE) endif() +if(ENABLE_PARANOID) + add_definitions(-DENABLE_PARANOID) +endif() + if(ENABLE_MULTITHREADING) add_definitions(-DHAVE_MULTITHREADING) endif() +if(ENABLE_INSECURE_AUTH_FOR_DEVTEST) + add_definitions(-DENABLE_INSECURE_AUTH_FOR_DEVTEST) +endif() + include(cmake/Macros.cmake) if(IOS) diff --git a/include/libnfs-private.h b/include/libnfs-private.h index a237421d..5e7272c4 100644 --- a/include/libnfs-private.h +++ b/include/libnfs-private.h @@ -133,15 +133,21 @@ struct rpc_fragment { /* * Queue is singly-linked but we hold on to the tail + * Using tailp this can be used to queue high and low priority pdus where high + * priority pdus are at the head of the queue while low priority pdus are + * queued behind the high priority pdus. tailp is the tail of high priority + * pdus, after which the low priority pdus start. */ struct rpc_queue { - struct rpc_pdu *head, *tail; + struct rpc_pdu *head, *tail, *tailp; }; #define DEFAULT_HASHES 4 #define NFS_RA_TIMEOUT 5 #define NFS_MIN_XFER_SIZE NFSMAXDATA2 -#define NFS_MAX_XFER_SIZE (4 * 1024 * 1024) +//#define NFS_MAX_XFER_SIZE (4 * 1024 * 1024) +/* 100MB MAX RPC size for supporting full size Blob block write */ +#define NFS_MAX_XFER_SIZE (100 * 1024 * 1024) #define NFS_DEF_XFER_SIZE (1 * 1024 * 1024) #define ZDR_ENCODE_OVERHEAD 1024 #define ZDR_ENCODEBUF_MINSIZE 4096 @@ -219,6 +225,13 @@ struct rpc_iovec_cursor { * At any point these many new bytes need to be read into this cursor. */ size_t remaining_size; + + /* + * Following ref are used to reset iov[] in case we need to resend + * this request, (possibly) after a reconnect. + */ + struct iovec *iov_ref; + int iovcnt_ref; }; enum input_state { @@ -232,8 +245,10 @@ enum input_state { #ifdef HAVE_TLS struct tls_cb_data { - rpc_cb cb; - void *private_data; +#define TLS_CB_DATA_MAGIC *((const uint32_t *) "TLCD") + uint32_t magic; + rpc_cb cb; + void *private_data; }; typedef enum tls_handshake_state { @@ -261,11 +276,67 @@ struct tls_context { #define INC_STATS(rpc, stat) ++((rpc)->stats.stat) +/** + * Auth related context information. + * It contains two types of information: + * - Information needed for querying the token to be used for auth. + * These are saved and read by the user and are opaue to libnfs. + * - Outcome of the auth process. + * These are used by libnfs. + */ +#define AUTH_CONTEXT_MAGIC *((const uint32_t *) "ACTX") + +struct auth_context { + uint32_t magic; + + /* /account/container for which the token is required */ + char *export_path; + + /* AuthType, currently only AzAuthAAD is supported */ + char *auth_type; + + /* Version of the client which initiates the auth request */ + char *client_version; + + /* ID of the client which initiates the auth request */ + char *client_id; + + + /* + * Is this connection successfully authorized? + * Updated after a successful call to get_token_callback_t. + * Cleared on token expiry. + */ + bool_t is_authorized; + + /* + * Does the token need to be refreshed? + * This is edge trigerred. It's set once when we discover that the + * current token has expired and then cleared once we setup reconnect + * which will eventually refresh the token. + */ + bool_t needs_refresh; + + /* + * Expiry time of the current token. + * Updated after a successful call to get_token_callback_t. + */ + uint64_t expiry_time; +}; + +struct azauth_cb_data { +#define AZAUTH_CB_DATA_MAGIC *((const uint32_t *) "AZCD") + uint32_t magic; + rpc_cb cb; + void *private_data; +}; + struct gss_ctx_id_struct; struct rpc_context { uint32_t magic; int fd; int old_fd; + int evfd; int is_connected; int is_nonblocking; @@ -277,18 +348,42 @@ struct rpc_context { struct AUTH *auth; uint32_t xid; + /* + * Queue of to-be-transmitted PDUs. + * Note: The PDU at the head of this queue will be the next one to be + * written to the socket. This can be a half-sent PDU for which + * (out.num_done < out.total_size). This implies that it's never + * safe to add anything to the head of this queue as that might + * cause the next rpc_write_to_socket() to incorrectly pick data + * from this new PDU while the previous one is half written. + * Only rpc_reconnect_requeue() can safely add to the head of + * this queue as it resets the connection and also the read and + * write cursors. Always use rpc_return_to_outqueue() to safely + * return a pdu to outqueue for retransmit. + */ struct rpc_queue outqueue; struct sockaddr_storage udp_src; uint32_t num_hashes; + + /* + * Queue of transmitted-and-awaiting-response PDUs. + */ struct rpc_queue *waitpdu; uint32_t waitpdu_len; uint32_t max_waitpdu_len; + /* + * Linux thread id returned by gettid(). + * Used for logging. + */ + pid_t tid; #ifdef HAVE_MULTITHREADING - int multithreading_enabled; libnfs_mutex_t rpc_mutex; #ifndef HAVE_STDATOMIC_H + int multithreading_enabled; libnfs_mutex_t atomic_int_mutex; +#else + atomic_int multithreading_enabled; #endif /* HAVE_STDATOMIC_H */ #endif /* HAVE_MULTITHREADING */ @@ -311,6 +406,21 @@ struct rpc_context { int auto_reconnect; int num_retries; + /* + * If true, reconnect will resolve 'server' afresh before reconnecting, + * else it'll reconnect to the last resolved address stored in + * rpc_context->s. + * Defaults to false and can be set by calling + * rpc_set_resolve_on_reconnect(). Once set it'll remain set for the + * life of the rpc transport and will decide the reconnect behaviour + * everytime a reconnect is needed. + * If resolve_on_reconnect is set, rpc_reconnect_requeue() will set + * resolve_server before calling rpc_connect_sockaddr_async() which + * will then resolve 'server' address before reconnecting. + */ + bool_t resolve_on_reconnect; + bool_t resolve_server; + /* * NFS server name or IP address. It has the following uses: * - Used for certificate verification, in case of xprtsec=[tls,mtls] @@ -323,6 +433,15 @@ struct rpc_context { */ char *server; + /* + * rpc_set_sockaddr() stores the same port number which it saves in + * rpc_context->s, so that it can be used during reconnect, so this is + * the last port to which this rpc_context connected. Note that an + * rpc_context may first connect to portmap and then to mount and then + * to nfs, all on different ports. + */ + int port; + /* fragment reassembly */ struct rpc_fragment *fragments; @@ -399,6 +518,20 @@ struct rpc_context { /* Context used for performing TLS handshake with the server */ struct tls_context tls_context; + + /* + * Do we need to perform auth on connect/reconnect? + * This starts as FALSE and is set to TRUE if user calls + * nfs_set_auth_context() to convey his intent to use auth for this + * rpc_context. + * If use_azauth is TRUE then a connection must send AZAUTH RPC as + * the very first RPC, to authn+authz the client with the server. + * If auth fails, no RPCs can be sent over the connection. + * If use_azauth is TRUE auth_context contains information needed for + * authn and authz and also holds the outcome of authn and authz. + */ + bool_t use_azauth; + struct auth_context auth_context; #endif /* HAVE_TLS */ #ifdef HAVE_LIBKRB5 @@ -430,7 +563,76 @@ struct rpc_pdu { int free_zdr; int free_pdu; - struct rpc_data outdata; +#ifdef ENABLE_PARANOID + #define PDU_PRESENT 0x05050505 + #define PDU_ABSENT 0xaaaaaaaa + /* + * We maintain some extra state inside pdu for performing paranoid + * checks. + */ + int added_to_outqueue_at_line; + uint64_t added_to_outqueue_at_time; + int removed_from_outqueue_at_line; + uint64_t removed_from_outqueue_at_time; + uint32_t in_outqueue; + + int added_to_waitpdu_at_line; + uint64_t added_to_waitpdu_at_time; + int removed_from_waitpdu_at_line; + uint64_t removed_from_waitpdu_at_time; + uint32_t in_waitpdu; +#endif + + /* + * Queueing priority that can be passed to rpc_queue_pdu2(). + * These have the following meaning: + * PDU_Q_PRIO_LOW - PDU will be queued at rpc_context.outqueue.tail. + * This adds the pdu behind all queued pdus. + * PDU_Q_PRIO_HI - PDU will be queued at rpc_context.outqueue.tailp. + * This adds the pdu to the tail of the high prio + * queue, behind already queued high prio pdus but + * ahead of all already queued low prio pdus. + * PDUs queued with PDU_Q_PRIO_HI will have + * is_high_prio set. + * PDU_Q_PRIO_HEAD - PDU will be queued at rpc_context.outqueue.head. + * This adds the pdu ahead of all queued pdus. + * PDUs queued with PDU_Q_PRIO_HEAD will have + * both is_head_prio and is_high_prio set. + */ + #define PDU_Q_PRIO_LOW 0 + #define PDU_Q_PRIO_HI 1 + #define PDU_Q_PRIO_HEAD 2 + + /* + * Is it a high-prio pdu, added by rpc_add_to_outqueue_highp()? + */ + bool_t is_high_prio; + + /* + * Is it a head-prio pdu, currently used only by AzAuth RPC. + * If this is TRUE, is_high_prio will also be TRUE, since head prio + * pdu is a high priority pdu. This is done for proper updation of + * various outqueue pointers. + */ + bool_t is_head_prio; + + /* + * Was this PDU retransmitted? + * libnfs lets its users know if a PDU that completed, was retransmitted + * or was it only sent to the server once. Users can use this info to + * do useful things, f.e., one of the thing they can do is work around + * the weakly consistent nature of NFS by treating an NFS3ERR_NOENT + * returned by a REMOVE/RMDIR call as NFS3_OK since it may have been + * deleted the first time it was sent and the subsequent retransmit + * may have gone to another node (which doesn't share the DRC cache) + * and hence it failed it with NFS3ERR_NOENT. + * Note that most applications will handle an unlink() call succeeding + * for a non-existent file better than unlink() call failing with + * NOENT for a file that was actually present. + */ + bool_t is_retransmitted; + + struct rpc_data outdata; /* For sending/receiving * out contains at least three vectors: @@ -458,6 +660,39 @@ struct rpc_pdu { * Used to clamp long reads. */ + /* + * Total request bytes sent out for this PDU. + * This includes RPC header + NFS header + optional data (for WRITE). + * This can be queried using rpc_pdu_get_req_size(pdu) after the + * rpc__ API returns the to-be-sent PDU. + * This can be used by applications that want to provide mountstats + * style "avg bytes sent" telemetry. + */ + uint32_t req_size; + + /* + * Total response bytes received for this PDU. + * This includes RPC header + NFS header + optional data (for READ). + * This can be queried using rpc_pdu_get_resp_size(rpc_get_pdu(rpc)) + * inside the rpc__ callback. + * This can be used by applications that want to provide mountstats + * style "avg bytes received" telemetry. + */ + uint32_t resp_size; + +#ifdef HAVE_CLOCK_GETTIME + /* + * Microseconds since epoch when this PDU was completely written to + * the socket. Note that due to TCP connection b/w and sndbuf size + * limitations this time can be very different from the time the PDU + * was queued to rpc->outqueue for sending, using rpc_queue_pdu(). + * Applications can use this to find the "rtt taken by the server to + * execute this RPC" by diff'ing this with the time when the callback + * is called. + */ + uint64_t dispatch_usecs; +#endif + rpc_cb cb; void *private_data; @@ -538,7 +773,12 @@ struct rpc_pdu { void rpc_reset_queue(struct rpc_queue *q); void rpc_enqueue(struct rpc_queue *q, struct rpc_pdu *pdu); -void rpc_return_to_queue(struct rpc_queue *q, struct rpc_pdu *pdu); +void rpc_add_to_outqueue_head(struct rpc_context *rpc, struct rpc_pdu *pdu); +void rpc_add_to_outqueue_headp(struct rpc_context *rpc, struct rpc_pdu *pdu); +void rpc_add_to_outqueue_highp(struct rpc_context *rpc, struct rpc_pdu *pdu); +void rpc_add_to_outqueue_lowp(struct rpc_context *rpc, struct rpc_pdu *pdu); +void rpc_return_to_outqueue(struct rpc_context *rpc, struct rpc_pdu *pdu); +int rpc_remove_pdu_from_queue(struct rpc_queue *q, struct rpc_pdu *remove_pdu); unsigned int rpc_hash_xid(struct rpc_context *rpc, uint32_t xid); struct rpc_pdu *rpc_allocate_pdu(struct rpc_context *rpc, int program, int version, int procedure, rpc_cb cb, void *private_data, zdrproc_t zdr_decode_fn, int zdr_bufsize); struct rpc_pdu *rpc_allocate_pdu2(struct rpc_context *rpc, int program, int version, int procedure, rpc_cb cb, void *private_data, zdrproc_t zdr_decode_fn, int zdr_bufsize, size_t alloc_hint, int iovcnt_hint); @@ -546,10 +786,15 @@ void pdu_set_timeout(struct rpc_context *rpc, struct rpc_pdu *pdu, uint64_t now_ void rpc_free_pdu(struct rpc_context *rpc, struct rpc_pdu *pdu); int rpc_queue_pdu(struct rpc_context *rpc, struct rpc_pdu *pdu); +int rpc_queue_pdu2(struct rpc_context *rpc, struct rpc_pdu *pdu, int prio); int rpc_process_pdu(struct rpc_context *rpc, char *buf, int size); struct rpc_pdu *rpc_find_pdu(struct rpc_context *rpc, uint32_t xid); void rpc_error_all_pdus(struct rpc_context *rpc, const char *error); +#ifdef ENABLE_PARANOID +void rpc_paranoid_checks(struct rpc_context *rpc); +#endif + /* * XXX This holds rpc->rpc_mutex, so if the caller is already holding * rpc->rpc_mutex, use the nolock version below. @@ -587,12 +832,22 @@ void nfs_set_error_locked(struct nfs_context *nfs, char *error_string, ...) #define RPC_LOG(rpc, level, format, ...) ; #define LOG(rpc, level, format, ...) ; #else +#ifdef HAVE_MULTITHREADING +#define RPC_LOG(rpc, level, format, ...) \ + do { \ + if (level <= rpc->debug) { \ + fprintf(stderr, "[%d] libnfs:%d rpc %p " format "\n", rpc->tid, level, rpc, ## __VA_ARGS__); \ + } \ + } while (0) +#else #define RPC_LOG(rpc, level, format, ...) \ do { \ if (level <= rpc->debug) { \ fprintf(stderr, "libnfs:%d rpc %p " format "\n", level, rpc, ## __VA_ARGS__); \ } \ } while (0) +#endif /* HAVE_MULTITHREADING */ + /* * Use LOG() for logging from code where there is no rpc_context. * It only provides simple unconditional logging since we don't have any debug @@ -633,6 +888,9 @@ int rpc_add_fragment(struct rpc_context *rpc, char *data, uint32_t size); void rpc_free_all_fragments(struct rpc_context *rpc); int rpc_is_udp_socket(struct rpc_context *rpc); uint64_t rpc_current_time(void); +#ifdef HAVE_CLOCK_GETTIME +uint64_t rpc_wallclock_time(void); +#endif void *zdr_malloc(ZDR *zdrs, uint32_t size); @@ -818,6 +1076,7 @@ void rpc_shrink_cursor(struct rpc_context *rpc, struct rpc_iovec_cursor *v, void rpc_memcpy_cursor(struct rpc_context *rpc, struct rpc_iovec_cursor *v, const void *src, size_t len); void rpc_free_cursor(struct rpc_context *rpc, struct rpc_iovec_cursor *v); +void rpc_reset_cursor(struct rpc_context *rpc, struct rpc_iovec_cursor *v); const struct nfs_fh *nfs_get_rootfh(struct nfs_context *nfs); int nfs_normalize_path(struct nfs_context *nfs, char *path); @@ -986,6 +1245,7 @@ int nfs4_write_async(struct nfs_context *nfs, struct nfsfh *nfsfh, void *private_data); int rpc_write_to_socket(struct rpc_context *rpc); +bool_t rpc_auth_needs_refresh(struct rpc_context *rpc); int _nfs_mount_async(struct nfs_context *nfs, const char *server, const char *exportname, nfs_cb cb, void *private_data); diff --git a/include/nfsc/libnfs-raw.h b/include/nfsc/libnfs-raw.h index 96218ef6..c7b5defd 100644 --- a/include/nfsc/libnfs-raw.h +++ b/include/nfsc/libnfs-raw.h @@ -87,9 +87,19 @@ struct rpc_stats { /* * RPC requests which didn't get a response for timeo period. * See mount option 'timeo'. + * These indicate some issue with the server and/or connection. */ uint64_t num_timedout; + /* + * RPC requests that timed out while sitting in outqueue. + * Unlike num_timedout, these are requests which were not sent to + * server. If this number is high it indicates a slow or unresponsive + * server and/or slow connection. Application should slow down issuing + * new RPC requests. + */ + uint64_t num_timedout_in_outqueue; + /* * RPC requests which didn't get a response even after retrans * retries. These are counted in num_timedout as well. @@ -109,6 +119,26 @@ struct rpc_stats { * - Major timeout was observed. */ uint64_t num_reconnects; + + /* + * rpc->outqueue length. + */ + uint32_t outqueue_len; + + /* + * rpc->waitpdu_len. + */ + uint32_t waitpdu_len; + + /* + * Stats for finding avg number of bytes we could write before + * writev() returned EAGAIN. This is an indication of the receive + * window advertised by the server. + */ + uint64_t last_write_bytes_before_eagain; + uint64_t tot_write_bytes_before_eagain; + uint64_t num_write_eagain; + }; struct rpc_context; @@ -162,9 +192,72 @@ EXTERN void rpc_set_auth(struct rpc_context *rpc, struct AUTH *auth); * your event system and passing revents as 0. */ EXTERN int rpc_get_fd(struct rpc_context *rpc); +EXTERN int rpc_get_evfd(struct rpc_context *rpc); EXTERN int rpc_which_events(struct rpc_context *rpc); EXTERN int rpc_service(struct rpc_context *rpc, int revents); +/* + * Returns the request PDU (rpc->pdu) inside a rpc_cb callback called after + * receiving the response from the server for a RPC request sent by us. + * User can pass this to rpc_pdu_get_resp_size() to find the total response + * size in bytes, which includes the RPC header, the NFS header and data bytes + * if any. + * + * Note: This can be legitimately called *only* inside the callback function + * where we are certain that the response was received, and hence + * rpc->pdu would be valid. Once the callback returns, rpc->pdu would + * be freed by libnfs so don't access the returned pdu outside the + * callback + */ +EXTERN struct rpc_pdu *rpc_get_pdu(struct rpc_context *rpc); + +/* + * Was this PDU retransmitted? i.e., did we send it to the server more than + * once? User can use this info to relax some errors, f.e., a REMOVE request + * that fails with NFS3ERR_NOENT can be treated as success if it was a + * retransmited request. + * Note that most applications will handle an unlink() call succeeding for a + * non-existent file better than unlink() call failing with NOENT for a file + * that was actually present. + */ +EXTERN bool_t rpc_pdu_is_retransmitted(struct rpc_pdu *pdu); + +/* + * Get the size in bytes of the RPC request that libnfs will send out for + * this PDU. The size includes the RPC header, the NFS header and any data + * bytes sent (only WRITE RPC request will send data bytes). + * The PDU passed to it would be the one returned by any of the rpc__ + * functions. + */ +EXTERN uint32_t rpc_pdu_get_req_size(struct rpc_pdu *pdu); + +/* + * Get the size in bytes of the RPC response that libnfs received for this + * request PDU. The size includes the RPC header, the NFS header and any data + * bytes received (only READ RPC response will contain data bytes). + * The PDU passed to it would be the one returned by rpc_get_pdu() called + * inside a callback function. + * + * Note: This can be legitimately called *only* inside a callback function. + * See rpc_get_pdu() for more details. + */ +EXTERN uint32_t rpc_pdu_get_resp_size(struct rpc_pdu *pdu); + +/* + * Get the wallclock time in microseconds since epoch when this request PDU + * was dispatched, i.e., fully sent out of the socket. Applications can use + * this to find out the time taken by the server to execute the PDU by finding + * the difference between this and the time when the response is received, and + * the callback is called. + * The PDU passed to it would be the one returned by rpc_get_pdu() called + * inside a callback function. + * + * Note: This can be legitimately called *only* inside a callback function. + * See rpc_get_pdu() for more details. + * Note: This only works on systems with clock_gettime() defined, else it + * returns 0. + */ +EXTERN uint64_t rpc_pdu_get_dispatch_usecs(struct rpc_pdu *pdu); /* * Returns the number of commands in-flight. Can be used by the application @@ -175,6 +268,13 @@ EXTERN int rpc_service(struct rpc_context *rpc, int revents); */ EXTERN int rpc_queue_length(struct rpc_context *rpc); +/* + * Call this if you want the server name to be resolved before reconnect. + * Default behaviour is to not resolve on reconnect and instead connect + * to the address resolved on initial connect. + */ +EXTERN void rpc_set_resolve_on_reconnect(struct rpc_context *rpc); + /* * Returns the number of commands awaiting from the server. * Can be used by the application to check if there are any @@ -1216,6 +1316,12 @@ rpc_nfs3_commit_task(struct rpc_context *rpc, rpc_cb cb, struct COMMIT3args *args, void *private_data); +struct AZAUTH3args; +EXTERN struct rpc_pdu * +rpc_nfs3_azauth_task(struct rpc_context *rpc, rpc_cb cb, + struct AZAUTH3args *args, + void *private_data); + /* * Call NFS3/SETATTR * @@ -2639,7 +2745,31 @@ rpc_null_task(struct rpc_context *rpc, int program, int version, EXTERN struct rpc_pdu * rpc_null_task_authtls(struct rpc_context *rpc, int nfs_version, rpc_cb cb, void *private_data); -#endif +#endif /* HAVE_TLS */ + +/* + * Authenticate and authorize the client with the server. Since each connection + * is separately authenticated, this must be called for each connection. + * It calls the callback registered through set_auth_token_callback() for + * querying the token, sends this token (and other required parameters) to the + * server using an AZAUTH RPC and updates rpc->auth_context as per the result + * of the auth returned by the server. + * + * Function returns + * pdu : The command was queued successfully. The callback will be invoked once + * the command completes. + * NULL : An error occured when trying to queue the command. + * The callback will not be invoked. + * + * When the callback is invoked, status indicates the result: + * RPC_STATUS_SUCCESS : We got a successful response from the server. + * data is NULL. + * RPC_STATUS_ERROR : The command failed with an error. + * RPC_STATUS_CANCEL : The command was cancelled. + * data is NULL. + */ +EXTERN struct rpc_pdu * +rpc_perform_azauth(struct rpc_context *rpc, rpc_cb cb, void *private_data); #ifdef __cplusplus } diff --git a/include/nfsc/libnfs.h b/include/nfsc/libnfs.h index 9f8ccac2..2f4f422e 100755 --- a/include/nfsc/libnfs.h +++ b/include/nfsc/libnfs.h @@ -48,6 +48,30 @@ extern "C" { struct nfs_context; struct rpc_context; +struct auth_context; +struct AZAUTH3args; + +/** + * Auth token info returned by get_token_callback_t. + */ +struct auth_token_cb_res { + /* + * This is the auth token set by the caller. + * It is a json string containing token and other relevant data + * sent as-is to the server in the AZAUTH3args.authdata arg of AzAuth + * RPC. + */ + char *azauth_data; + + /* + * Expiry time of the token contained in azauth_data. + * It is in seconds since unix epoch. + * libnfs will save this in auth_context.expiry_time and use it to + * correctly refresh the token before it expires. + */ + uint64_t expiry_time; +}; + struct nfs_url { char *server; char *path; @@ -96,6 +120,9 @@ struct utimbuf { * Used for interfacing the async version of the api into an external * eventsystem. * + * nfs_get_tid() returns the Linux tid of the libnfs thread processing + * requests for this nfs_context. + * * nfs_get_fd() returns the file descriptor for the context we need to * listen for events from. * @@ -120,7 +147,9 @@ struct utimbuf { * You only need this for the async interface. The sync interface already * do this in their built-in event loops. */ +EXTERN int nfs_get_tid(struct nfs_context *nfs); EXTERN int nfs_get_fd(struct nfs_context *nfs); +EXTERN int nfs_get_evfd(struct nfs_context *nfs); EXTERN int nfs_which_events(struct nfs_context *nfs); EXTERN int nfs_service(struct nfs_context *nfs, int revents); @@ -131,6 +160,13 @@ EXTERN int nfs_service(struct nfs_context *nfs, int revents); */ EXTERN int nfs_queue_length(struct nfs_context *nfs); +/* + * Call this if you want the server name to be resolved before reconnect. + * Default behaviour is to not resolve on reconnect and instead connect + * to the address resolved on initial connect. + */ +EXTERN void nfs_set_resolve_on_reconnect(struct nfs_context *nfs); + /* * Used if you need different credentials than the default for the current user. */ @@ -147,7 +183,7 @@ enum rpc_sec { RPC_SEC_KRB5P, }; EXTERN void nfs_set_security(struct nfs_context *nfs, enum rpc_sec sec); - + #ifdef HAVE_TLS /* * Various transport level security values that map to the mount option @@ -204,6 +240,19 @@ EXTERN struct nfs_context *nfs_init_context(void); */ EXTERN void nfs_destroy_context(struct nfs_context *nfs); +/* + * Function pointer type for getting auth token. + * It takes the auth_context, fetches the token for this context i.e., for the tenant, + * performs sanity checks and prepares the AZAUTH3args to be passed to rpc_nfs3_azauth_task. + * It returns auth_token_cb_res containing AZAUTH3args and expiry time of the token fetched. + */ +typedef struct auth_token_cb_res *(*get_token_callback_t)(struct auth_context *auth); + +/* + * Function to set get_token_callback_t. + */ +EXTERN void set_auth_token_callback(get_token_callback_t get_cb); + /* * Commands that are in flight are kept on linked lists and keyed by * XID so that responses received can be matched with a request. @@ -222,7 +271,6 @@ EXTERN void nfs_destroy_context(struct nfs_context *nfs); */ EXTERN int nfs_set_hash_size(struct nfs_context *nfs, int hashes); - /* * URL parsing functions. * These functions all parse a URL of the form @@ -280,6 +328,17 @@ EXTERN int nfs_set_hash_size(struct nfs_context *nfs, int hashes); EXTERN struct nfs_url *nfs_parse_url_full(struct nfs_context *nfs, const char *url); +/* + * Used to set values in auth_context present in nfs->rpc. + * User should call this function IFF they want to enable auth for the given + * nfs_context. + */ +EXTERN int nfs_set_auth_context(struct nfs_context *nfs, + const char *export_path, + const char *authtype, + const char *client_version, + const char *client_id); + /* * Parse an NFS URL, but do not split path and file. File * in the resulting struct remains NULL. @@ -327,7 +386,7 @@ EXTERN void nfs_set_readmax(struct nfs_context *nfs, size_t readmax); * Set the maximum supported WRITE size by the server */ EXTERN void nfs_set_writemax(struct nfs_context *nfs, size_t writemax); - + /* * MODIFY CONNECT PARAMETERS */ @@ -2031,18 +2090,25 @@ EXTERN void nfs4_set_verifier(struct nfs_context *nfs, const char *verifier); /* * MULTITHREADING - */ + */ /* * This function starts a separate service thread for multithreading support. * When multithreading is enabled the eventdriven async API is no longer * supported and you can only use the synchronous API. + * + * nfs_mt_service_thread_start_ss() is same as nfs_mt_service_thread_start() + * but allows caller to set the stack size of the libnfs service thread. + * Default 8MB stack size is not sufficient for very large readdir/readdirplus + * responses as the zdr decoder is recursive. */ -EXTERN int nfs_mt_service_thread_start(struct nfs_context *nfs); +EXTERN int nfs_mt_service_thread_start_ss(struct nfs_context *nfs, size_t stack_bytes); +#define nfs_mt_service_thread_start(nfs) nfs_mt_service_thread_start_ss(nfs, 0) + /* * Shutdown multithreading support. */ EXTERN void nfs_mt_service_thread_stop(struct nfs_context *nfs); - + #ifdef __cplusplus } #endif diff --git a/lib/init.c b/lib/init.c index 87ffa82a..b4d10f05 100644 --- a/lib/init.c +++ b/lib/init.c @@ -54,6 +54,8 @@ #include #include #include +#include + #include "slist.h" #include "libnfs-zdr.h" #include "libnfs.h" @@ -70,6 +72,9 @@ static const char *oom = "out of memory"; +/* + * This returns time in millseconds since system boot. + */ uint64_t rpc_current_time(void) { #ifdef HAVE_CLOCK_GETTIME @@ -82,6 +87,19 @@ uint64_t rpc_current_time(void) #endif } +#ifdef HAVE_CLOCK_GETTIME +/* + * This returns time in microseconds since epoch. + */ +uint64_t rpc_wallclock_time(void) +{ + struct timespec tp; + + clock_gettime(CLOCK_REALTIME, &tp); + return (uint64_t)tp.tv_sec * 1000000 + tp.tv_nsec / 1000; +} +#endif + int rpc_set_hash_size(struct rpc_context *rpc, int hashes) { uint32_t i; @@ -138,8 +156,20 @@ struct rpc_context *rpc_init_context(void) #endif #endif /* HAVE_MULTITHREADING */ + /* + * eventfd for informing RPC service thread of new PDUs waiting to + * be sent. + */ + rpc->evfd = eventfd(0, EFD_NONBLOCK); + if (rpc->evfd == -1) { + free(rpc->waitpdu); + free(rpc); + return NULL; + } + rpc->auth = authunix_create_default(); if (rpc->auth == NULL) { + close(rpc->evfd); free(rpc->waitpdu); free(rpc); return NULL; @@ -158,6 +188,7 @@ struct rpc_context *rpc_init_context(void) rpc->gid = getgid(); #endif rpc_reset_queue(&rpc->outqueue); + assert(rpc->stats.outqueue_len == 0); /* Default is no limit */ rpc->max_waitpdu_len = 0; @@ -376,6 +407,13 @@ char *rpc_get_error(struct rpc_context *rpc) return rpc->error_string ? rpc->error_string : ""; } +struct rpc_pdu *rpc_get_pdu(struct rpc_context *rpc) +{ + assert(rpc->magic == RPC_CONTEXT_MAGIC); + + return rpc->pdu; +} + void rpc_get_stats(struct rpc_context *rpc, struct rpc_stats *stats) { assert(rpc->magic == RPC_CONTEXT_MAGIC); @@ -386,6 +424,11 @@ void rpc_get_stats(struct rpc_context *rpc, struct rpc_stats *stats) } #endif /* HAVE_MULTITHREADING */ +#ifdef ENABLE_PARANOID + rpc_paranoid_checks(rpc); +#endif + + rpc->stats.waitpdu_len = rpc->waitpdu_len; *stats = rpc->stats; #ifdef HAVE_MULTITHREADING @@ -423,9 +466,18 @@ static void rpc_purge_all_pdus(struct rpc_context *rpc, int status, const char * while ((pdu = outqueue.head) != NULL) { outqueue.head = pdu->next; pdu->next = NULL; +#ifdef ENABLE_PARANOID + assert(pdu->in_outqueue == PDU_PRESENT); + assert(pdu->in_waitpdu == PDU_ABSENT); + pdu->in_outqueue = PDU_ABSENT; +#endif pdu->cb(rpc, status, (void *) error, pdu->private_data); rpc_free_pdu(rpc, pdu); + + assert(rpc->stats.outqueue_len > 0); + rpc->stats.outqueue_len--; } + assert(rpc->stats.outqueue_len == 0); #ifdef HAVE_MULTITHREADING if (rpc->multithreading_enabled) { nfs_mt_mutex_unlock(&rpc->rpc_mutex); @@ -450,6 +502,11 @@ static void rpc_purge_all_pdus(struct rpc_context *rpc, int status, const char * while((pdu = waitqueue.head) != NULL) { waitqueue.head = pdu->next; pdu->next = NULL; +#ifdef ENABLE_PARANOID + assert(pdu->in_outqueue == PDU_ABSENT); + assert(pdu->in_waitpdu == PDU_PRESENT); + pdu->in_waitpdu = PDU_ABSENT; +#endif pdu->cb(rpc, status, (void *) error, pdu->private_data); rpc_free_pdu(rpc, pdu); } @@ -519,10 +576,24 @@ void rpc_destroy_context(struct rpc_context *rpc) rpc->auth =NULL; } + /* + * Free pointers inside rpc->auth_context. + */ + free(rpc->auth_context.export_path); + free(rpc->auth_context.auth_type); + free(rpc->auth_context.client_version); + free(rpc->auth_context.client_id); + rpc->auth_context.is_authorized = FALSE; + rpc->use_azauth = FALSE; + if (rpc->fd != -1) { close(rpc->fd); } + if (rpc->evfd != -1) { + close(rpc->evfd); + } + if (rpc->error_string && rpc->error_string != oom) { free(rpc->error_string); rpc->error_string = NULL; @@ -674,6 +745,10 @@ void rpc_advance_cursor(struct rpc_context *rpc, struct rpc_iovec_cursor *v, /* remaining_size can only be 0 when iovcnt is 0 and v.v. */ assert((v->iovcnt == 0) == (v->remaining_size == 0)); + assert(v->iovcnt <= v->iovcnt_ref); + assert(v->iov >= v->base); + assert(v->iov <= v->iov_ref); + assert(v->iov_ref == (v->base + v->iovcnt_ref)); } /* @@ -705,6 +780,10 @@ void rpc_shrink_cursor(struct rpc_context *rpc, struct rpc_iovec_cursor *v, /* remaining_size can only be 0 when iovcnt is 0 and v.v. */ assert((v->iovcnt == 0) == (v->remaining_size == 0)); + assert(v->iovcnt <= v->iovcnt_ref); + assert(v->iov >= v->base); + assert(v->iov <= v->iov_ref); + assert(v->iov_ref == (v->base + v->iovcnt_ref)); } /* @@ -736,6 +815,33 @@ void rpc_memcpy_cursor(struct rpc_context *rpc, struct rpc_iovec_cursor *v, /* remaining_size can only be 0 when iovcnt is 0 and v.v. */ assert((v->iovcnt == 0) == (v->remaining_size == 0)); + assert(v->iovcnt <= v->iovcnt_ref); + assert(v->iov >= v->base); + assert(v->iov <= v->iov_ref); + assert(v->iov_ref == (v->base + v->iovcnt_ref)); +} + +void rpc_reset_cursor(struct rpc_context *rpc, struct rpc_iovec_cursor *v) +{ + int i; + + if (!v->base) { + return; + } + + assert(v->iovcnt <= v->iovcnt_ref); + assert(v->iov >= v->base); + assert(v->iov <= v->iov_ref); + assert(v->iov_ref == (v->base + v->iovcnt_ref)); + + v->iovcnt = v->iovcnt_ref; + v->iov = v->base; + + v->remaining_size = 0; + for (i = 0; i < v->iovcnt_ref; i++) { + v->iov[i] = v->iov_ref[i]; + v->remaining_size += v->iov[i].iov_len; + } } void rpc_free_cursor(struct rpc_context *rpc, struct rpc_iovec_cursor *v) diff --git a/lib/libnfs.c b/lib/libnfs.c index fea3f36c..bbc11024 100755 --- a/lib/libnfs.c +++ b/lib/libnfs.c @@ -193,18 +193,72 @@ nfs_set_xprtsecurity(struct nfs_context *nfs, enum rpc_xprtsec xprtsec) } #endif /* HAVE_TLS */ +int +nfs_get_tid(struct nfs_context *nfs) +{ + assert(nfs->rpc->magic == RPC_CONTEXT_MAGIC); + return nfs->rpc->tid; +} + int nfs_get_fd(struct nfs_context *nfs) { return rpc_get_fd(nfs->rpc); } +int +nfs_get_evfd(struct nfs_context *nfs) +{ + return rpc_get_evfd(nfs->rpc); +} + int nfs_queue_length(struct nfs_context *nfs) { return rpc_queue_length(nfs->rpc); } +void +nfs_set_resolve_on_reconnect(struct nfs_context *nfs) +{ + rpc_set_resolve_on_reconnect(nfs->rpc); +} + +/* Static variables to hold the registered get/put token callbacks */ +static get_token_callback_t get_auth_token_cb = NULL; + +void +set_auth_token_callback(get_token_callback_t get_cb) +{ + assert(get_cb); + + get_auth_token_cb = get_cb; +} + +static struct auth_token_cb_res * +get_azauth_token(struct auth_context *auth) +{ + return get_auth_token_cb(auth); +} + +static void +free_azauth_token(struct auth_token_cb_res *res, AZAUTH3args *args) +{ + /* Caller MUST pass both valid pointers */ + assert(res && args); + + free(args->client_version); + free(args->clientid); + free(args->authtype); + free(args->authtarget); + free(args->authdata); + + free(res->azauth_data); + + /* Free the auth_token_cb_res structure */ + free(res); +} + int nfs_which_events(struct nfs_context *nfs) { @@ -534,6 +588,46 @@ nfs_parse_url(struct nfs_context *nfs, const char *url, int dir, int incomplete) return urls; } +int nfs_set_auth_context(struct nfs_context *nfs, + const char *export_path, + const char *authtype, + const char *client_version, + const char *client_id) +{ + assert(export_path); + assert(authtype); + assert(client_version); + assert(client_id); + + if (nfs->rpc) { +#ifndef ENABLE_INSECURE_AUTH_FOR_DEVTEST + /* + * If not devtest, don't allow auth unless transport is secure. + */ + if (nfs->rpc->wanted_xprtsec == RPC_XPRTSEC_NONE) { + RPC_LOG(nfs->rpc, 1, "Cannot enable auth for xprtsec=none"); + return -1; + } +#endif + assert(nfs->rpc->use_azauth == FALSE); + + if (!strcmp(authtype,"AzAuthAAD")) { + nfs->rpc->use_azauth = TRUE; + } + + nfs->rpc->auth_context.magic = AUTH_CONTEXT_MAGIC; + nfs->rpc->auth_context.export_path = strdup(export_path); + nfs->rpc->auth_context.auth_type = strdup(authtype); + nfs->rpc->auth_context.client_version = strdup(client_version); + nfs->rpc->auth_context.client_id = strdup(client_id); + nfs->rpc->auth_context.is_authorized = FALSE; + nfs->rpc->auth_context.expiry_time = 0; + + return 0; + } + return -1; +} + struct nfs_url * nfs_parse_url_full(struct nfs_context *nfs, const char *url) { @@ -788,10 +882,91 @@ rpc_connect_program_6_cb(struct rpc_context *rpc, int status, } #endif /* HAVE_LIBKRB5 */ +void free_azauth_cb_data(struct azauth_cb_data *data) +{ + assert(data->magic == AZAUTH_CB_DATA_MAGIC); + free(data); +} + #ifdef HAVE_TLS void free_tls_cb_data(struct tls_cb_data *data) { - free(data); + assert(data->magic == TLS_CB_DATA_MAGIC); + free(data); +} + +/* + * Callback function called when we get a response for an AZAUTH RPC from the + * server. + * On a successful response, confirming the token is valid, we proceed with + * the next step of mount. + */ +static void +rpc_connect_program_4_2_cb(struct rpc_context *rpc, int status, + void *command_data, void *private_data) +{ + struct azauth_cb_data *data = private_data; + + assert(data->magic == AZAUTH_CB_DATA_MAGIC); + + assert(rpc->magic == RPC_CONTEXT_MAGIC); + /* Must be called only when use_azauth is true */ + assert(rpc->use_azauth); + /* rpc_perform_azauth() MUST have set is_authorized to FALSE */ + assert(rpc->auth_context.is_authorized == FALSE); + + if (status != RPC_STATUS_SUCCESS) { + RPC_LOG(rpc, 1, "AZAUTH RPC failure, status = %d", status); + + data->cb(rpc, status, command_data, data->private_data); + free_azauth_cb_data(data); + return; + } + + /* + * For successful AZAUTH, command_data should contain AZAUTH3res as + * returned by the server. + */ + assert(command_data); + + const AZAUTH3res *const res = command_data; + + /* Must be a valid NFS status */ + assert(nfsstat3_to_errno(res->status) != -ERANGE); + + if (res->status != NFS3_OK) { + RPC_LOG(rpc, 1, "AZAUTH NFS failure, status = %s", + nfsstat3_to_str(res->status)); + + /* + * Caller doesn't care if it's NFS error or RPC error. + * For any failure we should call the callback with failed + * status. + */ + data->cb(rpc, RPC_STATUS_ERROR, NULL, data->private_data); + free_azauth_cb_data(data); + return; + } + + RPC_LOG(rpc, 2, "AZAUTH successful!"); + + const char *server_version = res->AZAUTH3res_u.resok.server_version; + const char *server_id = res->AZAUTH3res_u.resok.serverid; + + assert(server_version); + assert(server_id); + + RPC_LOG(rpc, 2, "AZAUTH Server version=%s Served id=%s", + server_version, server_id); + + /* AZAUTH RPC successful, connection is now authorized */ + rpc->auth_context.is_authorized = TRUE; + + /* + * Now that AZAUTH is successful, call the next in chain. + */ + data->cb(rpc, RPC_STATUS_SUCCESS, NULL, data->private_data); + free_azauth_cb_data(data); } /* @@ -806,6 +981,7 @@ rpc_connect_program_4_1_cb(struct rpc_context *rpc, int status, { struct tls_cb_data *data = private_data; + assert(data->magic == TLS_CB_DATA_MAGIC); assert(rpc->magic == RPC_CONTEXT_MAGIC); RPC_LOG(rpc, 2, "Got AUTH_TLS response, status=%d", status); @@ -835,7 +1011,7 @@ rpc_connect_program_4_1_cb(struct rpc_context *rpc, int status, case TLS_HANDSHAKE_COMPLETED: RPC_LOG(rpc, 2, "do_tls_handshake: TLS handshake completed " "synchronously on fd %d", rpc->fd); - data->cb(rpc, RPC_STATUS_SUCCESS, NULL, data->private_data); + data->cb(rpc, RPC_STATUS_SUCCESS, NULL, data->private_data); break; case TLS_HANDSHAKE_FAILED: RPC_LOG(rpc, 2, "do_tls_handshake: Failed to start TLS handshake, or " @@ -909,6 +1085,47 @@ rpc_connect_program_5_cb(struct rpc_context *rpc, int status, free_rpc_cb_data(data); } +static void +rpc_connect_program_5_0_cb(struct rpc_context *rpc, int status, + void *command_data, void *private_data) +{ + struct rpc_cb_data *data = private_data; + + assert(rpc->magic == RPC_CONTEXT_MAGIC); + + /* rpc_connect_program_5_0_cb() MUST be called only for TLS connections */ + assert(rpc->use_tls); + + /* Dont want any more callbacks even if the socket is closed */ + rpc->connect_cb = NULL; + + if (status != RPC_STATUS_SUCCESS) { + assert(rpc->tls_context.state != TLS_HANDSHAKE_COMPLETED); + + data->cb(rpc, status, command_data, data->private_data); + free_rpc_cb_data(data); + return; + } + + /* + * TLS handshake is completed. + * If use_azauth is set, perform azauth now. + */ + assert(rpc->tls_context.state == TLS_HANDSHAKE_COMPLETED); + + if (rpc->use_azauth) { + if (rpc_perform_azauth(rpc, rpc_connect_program_5_cb, + data) == NULL) { + data->cb(rpc, RPC_STATUS_ERROR, NULL, data->private_data); + free_rpc_cb_data(data); + return; + } + } else { + rpc_connect_program_5_cb(rpc, RPC_STATUS_SUCCESS, NULL, data); + return; + } +} + static void rpc_connect_program_4_cb(struct rpc_context *rpc, int status, void *command_data, void *private_data) @@ -924,7 +1141,7 @@ rpc_connect_program_4_cb(struct rpc_context *rpc, int status, data->cb(rpc, status, command_data, data->private_data); free_rpc_cb_data(data); return; - } + } #ifdef HAVE_TLS /* @@ -939,13 +1156,35 @@ rpc_connect_program_4_cb(struct rpc_context *rpc, int status, /* We should not use TLS for anything other than NFS */ assert(data->program == NFS_PROGRAM); + /* + * rpc_connect_program_5_0_cb() will be called when TLS + * handshake completes (success or failure). If TLS handshake + * is successful and use_azauth is TRUE it performs azauth + * and then calls data->cb() else calls data->cb() directly. + */ if (rpc_null_task_authtls(rpc, data->version, - rpc_connect_program_5_cb, data) == NULL) { + rpc_connect_program_5_0_cb, data) == NULL) { data->cb(rpc, RPC_STATUS_ERROR, command_data, data->private_data); free_rpc_cb_data(data); return; } } else +#endif /* HAVE_TLS */ + +#ifdef ENABLE_INSECURE_AUTH_FOR_DEVTEST + if (rpc->use_azauth) { + /* + * Insecure connection, if azauth is enabled perform auth. + * + * Note: THIS WOULD SEND THE TOKEN OVER AN INSECURE CONNECTION + * AND MUST ONLY BE USED IN DEVTEST ON TRUSTED NETWORKS. + */ + if (rpc_perform_azauth(rpc, rpc_connect_program_5_cb, data) == NULL) { + data->cb(rpc, RPC_STATUS_ERROR, command_data, data->private_data); + free_rpc_cb_data(data); + return; + } + } else #endif if (rpc_null_task(rpc, data->program, data->version, rpc_connect_program_5_cb, data) == NULL) { @@ -2558,6 +2797,7 @@ rpc_null_task_authtls(struct rpc_context *rpc, int nfs_version, rpc_cb cb, "for AUTH_TLS NULL call"); return NULL; } + data->magic = TLS_CB_DATA_MAGIC; data->cb = cb; data->private_data = private_data; @@ -2588,3 +2828,63 @@ rpc_null_task_authtls(struct rpc_context *rpc, int nfs_version, rpc_cb cb, return pdu; } #endif /* HAVE_TLS */ + +struct rpc_pdu * +rpc_perform_azauth(struct rpc_context *rpc, rpc_cb cb, void *private_data) +{ + assert(rpc->magic == RPC_CONTEXT_MAGIC); + + struct rpc_pdu *pdu; + struct auth_token_cb_res *res = get_azauth_token(&rpc->auth_context); + if (!res) { + RPC_LOG(rpc, 1, "get_azauth_token() failed!"); + return NULL; + } + + assert(res->azauth_data); + assert(res->expiry_time != 0); + assert(res->expiry_time >= time(NULL)); + + AZAUTH3args azauthargs; + + azauthargs.client_version = strdup(rpc->auth_context.client_version); + azauthargs.clientid = strdup(rpc->auth_context.client_id); + azauthargs.authtype = strdup(rpc->auth_context.auth_type); + azauthargs.authtarget = strdup(rpc->auth_context.export_path); + azauthargs.authdata = strdup(res->azauth_data); + + RPC_LOG(rpc, 2, "AZAuth3Args: client_version: %s, client_id: %s, " + "authtype: %s, authtarget: %s", + azauthargs.client_version, + azauthargs.clientid, + azauthargs.authtype, + azauthargs.authtarget); + + rpc->auth_context.is_authorized = FALSE; + rpc->auth_context.expiry_time = res->expiry_time; + + struct azauth_cb_data *data = calloc(1, sizeof(*data)); + if (data == NULL) { + free_azauth_token(res, &azauthargs); + rpc_set_error(rpc, "Out of memory. Failed to allocate azauth_cb_data"); + return NULL; + } + + data->magic = AZAUTH_CB_DATA_MAGIC; + data->cb = cb; + data->private_data = private_data; + + RPC_LOG(rpc, 2, "Sending AZAUTH RPC"); + + pdu = rpc_nfs3_azauth_task(rpc, rpc_connect_program_4_2_cb, + &azauthargs, data); + if (pdu == NULL) { + rpc_set_error(rpc, "AZAUTH RPC failed to set pdu"); + free_azauth_token(res, &azauthargs); + free_azauth_cb_data(data); + return NULL; + } + + free_azauth_token(res, &azauthargs); + return pdu; +} diff --git a/lib/multithreading.c b/lib/multithreading.c index d101afee..4c8b6268 100644 --- a/lib/multithreading.c +++ b/lib/multithreading.c @@ -47,6 +47,7 @@ #include #endif +#include #include #include "libnfs.h" #include "libnfs-raw.h" @@ -75,35 +76,71 @@ nfs_tid_t nfs_mt_get_tid(void) static void *nfs_mt_service_thread(void *arg) { struct nfs_context *nfs = (struct nfs_context *)arg; - struct pollfd pfd; + struct pollfd pfd[2]; int revents; int ret; - nfs->rpc->multithreading_enabled = 1; + nfs->rpc->tid = gettid(); + + /* Prevent compiler reordering by introducing a dependency */ + if (nfs->rpc->tid != 0) { + nfs->rpc->multithreading_enabled = 1; + } + + assert(nfs->rpc->multithreading_enabled); + + pfd[0].fd = nfs_get_evfd(nfs); + pfd[0].events = POLLIN; while (nfs->rpc->multithreading_enabled) { - pfd.fd = nfs_get_fd(nfs); - pfd.events = nfs_which_events(nfs); - pfd.revents = 0; + pfd[0].revents = 0; + pfd[1].fd = nfs_get_fd(nfs); + pfd[1].events = nfs_which_events(nfs); + pfd[1].revents = 0; - ret = poll(&pfd, 1, nfs->rpc->poll_timeout); + ret = poll(pfd, 2, nfs->rpc->poll_timeout); if (ret < 0) { - nfs_set_error(nfs, "Poll failed"); + nfs_set_error(nfs, "Poll failed: %s", strerror(errno)); revents = -1; } else { - revents = pfd.revents; + if (pfd[0].revents != 0) { + uint64_t evread; + [[maybe_unused]] ssize_t evbytes; + + assert(pfd[0].revents == POLLIN); + evbytes = read(pfd[0].fd, &evread, sizeof(evread)); + assert(evbytes == 8); + assert(evread > 0); + } + + revents = pfd[1].revents; } + + /* + * nfs_service() failing is an unusual condition, take a pause + * before retrying. + */ if (nfs_service(nfs, revents) < 0) { - if (revents != -1) - nfs_set_error(nfs, "nfs_service failed"); + nfs_set_error(nfs, "nfs_service failed, revents=0x%x", + revents); + RPC_LOG(nfs->rpc, 2, "Sleeping 5 secs, before retrying!"); + sleep(5); } } return NULL; } -int nfs_mt_service_thread_start(struct nfs_context *nfs) +int nfs_mt_service_thread_start_ss(struct nfs_context *nfs, size_t stack_bytes) { - if (pthread_create(&nfs->nfsi->service_thread, NULL, + pthread_attr_t attr; + + pthread_attr_init(&attr); + + if (stack_bytes != 0) { + pthread_attr_setstacksize(&attr, stack_bytes); + } + + if (pthread_create(&nfs->nfsi->service_thread, &attr, &nfs_mt_service_thread, nfs)) { nfs_set_error(nfs, "Failed to start service thread"); return -1; @@ -117,8 +154,14 @@ int nfs_mt_service_thread_start(struct nfs_context *nfs) void nfs_mt_service_thread_stop(struct nfs_context *nfs) { + uint64_t evwrite = 1; + [[maybe_unused]] ssize_t evbytes; + + /* Signal the service thread to stop */ nfs->rpc->multithreading_enabled = 0; - pthread_join(nfs->nfsi->service_thread, NULL); + evbytes = write(rpc_get_evfd(nfs->rpc), &evwrite, sizeof(evwrite)); + assert(evbytes == 8); + pthread_join(nfs->nfsi->service_thread, NULL); } /* diff --git a/lib/pdu.c b/lib/pdu.c index 714848c4..94fd5829 100644 --- a/lib/pdu.c +++ b/lib/pdu.c @@ -56,6 +56,7 @@ #include #include #include +#include #include "slist.h" #include "libnfs-zdr.h" #include "libnfs.h" @@ -70,6 +71,7 @@ void rpc_reset_queue(struct rpc_queue *q) { q->head = NULL; q->tail = NULL; + q->tailp = NULL; } /* @@ -77,23 +79,282 @@ void rpc_reset_queue(struct rpc_queue *q) */ void rpc_enqueue(struct rpc_queue *q, struct rpc_pdu *pdu) { - if (q->head == NULL) + if (q->head == NULL) { + assert(q->tail == NULL); q->head = pdu; - else + } else { + assert(pdu != q->head); + assert(pdu != q->tail); q->tail->next = pdu; + } q->tail = pdu; pdu->next = NULL; } +/** + * Add pdu to the head of outqueue. + * It tries to add pdu to the head but if the pdu at the head is partially + * written to the socket it adds pdu after that. + * We do that to not mix data from different pdus being sent on the socket. + */ +void rpc_add_to_outqueue_head(struct rpc_context *rpc, struct rpc_pdu *pdu) +{ + if (rpc->outqueue.head == NULL) { + assert(rpc->outqueue.tail == NULL); + assert(rpc->outqueue.tailp == NULL); + assert(rpc->stats.outqueue_len == 0); + + rpc->outqueue.head = rpc->outqueue.tail = pdu; + if (pdu->is_high_prio) + rpc->outqueue.tailp = pdu; + pdu->next = NULL; + } else { + if (rpc->outqueue.head == rpc->outqueue.tail) { + assert(rpc->stats.outqueue_len == 1); + assert(rpc->outqueue.head->next == NULL); + assert(rpc->outqueue.tail->next == NULL); + assert(!rpc->outqueue.tailp || + (rpc->outqueue.tailp == rpc->outqueue.tail)); + assert(pdu != rpc->outqueue.head); + } else { + assert(rpc->stats.outqueue_len > 1); + assert(rpc->outqueue.head->next != NULL); + assert(rpc->outqueue.tail->next == NULL); + assert(pdu != rpc->outqueue.head); + assert(pdu != rpc->outqueue.tail); + assert(pdu != rpc->outqueue.tailp); + } + + /* + * Add to the head if head pdu is not partially-sent, else add + * after that. + * If no high prio pdu queued and this one is high prio pdu, + * set tailp as well, also if added after head pdu and tailp + * points at head and pdu is high prio update tailp. + */ + if (rpc->outqueue.head->out.num_done == 0) { + pdu->next = rpc->outqueue.head; + rpc->outqueue.head = pdu; + if (pdu->is_high_prio && (rpc->outqueue.tailp == NULL)) + rpc->outqueue.tailp = pdu; + } else { + if (pdu->is_high_prio && + ((rpc->outqueue.tailp == NULL) || + (rpc->outqueue.tailp == rpc->outqueue.head))) + rpc->outqueue.tailp = pdu; + pdu->next = rpc->outqueue.head->next; + rpc->outqueue.head->next = pdu; + } + + if (pdu->next == NULL) + rpc->outqueue.tail = pdu; + } + + assert(rpc->outqueue.tail->next == NULL); + assert(rpc->outqueue.tail->is_high_prio == + (rpc->outqueue.tailp == rpc->outqueue.tail)); + + if (rpc->stats.outqueue_len++ == 0) { + /* + * If this is the first pdu added to an empty outqueue let the + * service thread know. + */ + uint64_t evwrite = 1; + [[maybe_unused]] ssize_t evbytes = + write(rpc_get_evfd(rpc), &evwrite, sizeof(evwrite)); + assert(evbytes == 8); + } +} + +/** + * Head priority pdu is a high prio pdu added to outqueue.head, ahead of all + * high (and low) prio pdus. + */ +void rpc_add_to_outqueue_headp(struct rpc_context *rpc, struct rpc_pdu *pdu) +{ + /* + * AZAUTH RPC is the only one queued with head priority and + * AZAUTH RPC MUST only be sent if use_azauth is true. + */ + assert(rpc->use_azauth); + + /* + * When rpc_add_to_outqueue_headp() is called there shouldn't be any + * partially sent pdu in the queue. It's typically called either when + * the connection is freshly created, at which time there are no pdus + * in outqueue, or on reconnect, at which time outqueue must have been + * reset and num_done must have been set to 0 for the head pdu. + */ + if (rpc->outqueue.head != NULL) { + assert(rpc->outqueue.head->out.num_done == 0); + } + + pdu->is_head_prio = TRUE; + pdu->is_high_prio = TRUE; + rpc_add_to_outqueue_head(rpc, pdu); + + assert(rpc->outqueue.head != NULL); + assert(rpc->outqueue.tail != NULL); + assert(rpc->outqueue.tailp != NULL); +} + +/** + * High priority pdus are added after tailp. + */ +void rpc_add_to_outqueue_highp(struct rpc_context *rpc, struct rpc_pdu *pdu) +{ + assert(pdu->is_head_prio == FALSE); + pdu->is_high_prio = TRUE; + if (rpc->outqueue.tailp == NULL) { + /* + * First high priority pdu, add to head. + */ + rpc_add_to_outqueue_head(rpc, pdu); + assert(rpc->outqueue.head != NULL); + assert(rpc->outqueue.tail != NULL); + assert(rpc->outqueue.tailp != NULL); + } else { + assert(rpc->outqueue.head != NULL); + assert(rpc->outqueue.tail != NULL); + assert(pdu != rpc->outqueue.head); + assert(pdu != rpc->outqueue.tail); + assert(pdu != rpc->outqueue.tailp); + assert(rpc->stats.outqueue_len > 0); + + pdu->next = rpc->outqueue.tailp->next; + rpc->outqueue.tailp->next = pdu; + if (rpc->outqueue.tail == rpc->outqueue.tailp) + rpc->outqueue.tail = pdu; + rpc->outqueue.tailp = pdu; + rpc->stats.outqueue_len++; + } +} + +/** + * Low priority pdus are added to the tail. + */ +void rpc_add_to_outqueue_lowp(struct rpc_context *rpc, struct rpc_pdu *pdu) +{ + assert(pdu->is_head_prio == FALSE); + pdu->is_high_prio = FALSE; + rpc_enqueue(&rpc->outqueue, pdu); + if (rpc->stats.outqueue_len++ == 0) { + /* + * If this is the first pdu added to an empty outqueue let the + * service thread know. + */ + uint64_t evwrite = 1; + [[maybe_unused]] ssize_t evbytes = + write(rpc_get_evfd(rpc), &evwrite, sizeof(evwrite)); + assert(evbytes == 8); + } + + assert(rpc->outqueue.head != NULL); + assert(rpc->outqueue.tail != NULL); + assert(pdu->next == NULL); +} + /* - * Push to the front/head of the queue + * Return pdu to outqueue to be retransmitted. + * It adds the pdu to the head of outqueue, unless the head pdu is partially + * sent, in which case it adds it right after the head pdu. */ -void rpc_return_to_queue(struct rpc_queue *q, struct rpc_pdu *pdu) +void rpc_return_to_outqueue(struct rpc_context *rpc, struct rpc_pdu *pdu) { - pdu->next = q->head; - q->head = pdu; - if (q->tail == NULL) - q->tail = pdu; + rpc_add_to_outqueue_head(rpc, pdu); + + /* + * Only already transmitted PDUs are added back to outqueue, so sending + * it out will entail a retransmit. + */ + INC_STATS(rpc, num_retransmitted); + pdu->is_retransmitted = 1; + + /* + * Reset output and input cursors as we have to re-send the whole pdu + * again (and read back the response fresh into pdu->in). + */ + pdu->out.num_done = 0; + rpc_reset_cursor(rpc, &pdu->in); +} + +/* + * Remove pdu from q. + * If found it'll remove the pdu and update q->head and q->tail correctly. + * Returns 0 if remove_pdu not found in q else returns 1. + */ +int rpc_remove_pdu_from_queue(struct rpc_queue *q, struct rpc_pdu *remove_pdu) +{ + if (q->head != NULL) { + struct rpc_pdu *pdu = q->head; + + assert(q->tail != NULL); + + /* + * remove_pdu is the head pdu. + * Change the head to point to the next pdu. + * If tail is also pointing to remove_pdu, this means it's the + * only PDU and after removing that we will have an empty list. + * If tailp is pointing to remove_pdu, this means it's the + * only high prio pdu and after removing it tailp will be NULL. + */ + if (q->head == remove_pdu) { + q->head = remove_pdu->next; + + if (q->tailp == remove_pdu) { + q->tailp = NULL; + } + + if (q->tail == remove_pdu) { + assert(remove_pdu->next == NULL); + q->tail = NULL; + assert(q->tailp == NULL); + assert(q->head == NULL); + } else { + assert(q->head != NULL); + } + + remove_pdu->next = NULL; + return 1; + } + + /* + * remove_pdu is not the head pdu. + * Search for it and if found, remove it, and update tail if + * tail is pointing to remove_pdu. + */ + while (pdu->next && pdu->next != remove_pdu) { + pdu = pdu->next; + } + + if (pdu->next == NULL) { + /* remove_pdu not found in q */ + return 0; + } + + pdu->next = remove_pdu->next; + + if (q->tail == remove_pdu) { + q->tail = pdu; + } + + if (q->tailp == remove_pdu) { + assert(remove_pdu->is_high_prio); + if (pdu->is_high_prio) + q->tailp = pdu; + else + q->tailp = NULL; + } + + remove_pdu->next = NULL; + + return 1; + } else { + assert(q->tail == NULL); + assert(q->tailp == NULL); + /* not found */ + return 0; + } } unsigned int rpc_hash_xid(struct rpc_context *rpc, uint32_t xid) @@ -182,6 +443,12 @@ struct rpc_pdu *rpc_allocate_pdu2(struct rpc_context *rpc, int program, int vers return NULL; } memset(pdu, 0, pdu_size); + +#ifdef ENABLE_PARANOID + /* PDU is not present in any queue to start with */ + pdu->in_outqueue = pdu->in_waitpdu = PDU_ABSENT; +#endif + #ifdef HAVE_MULTITHREADING if (rpc->multithreading_enabled) { nfs_mt_mutex_lock(&rpc->rpc_mutex); @@ -353,6 +620,9 @@ struct rpc_pdu *rpc_allocate_pdu2(struct rpc_context *rpc, int program, int vers rpc_add_iovector(rpc, &pdu->out, &pdu->outdata.data[4], zdr_getpos(&pdu->zdr), NULL); + /* Freshly allocated PDU cannot be retransmitted */ + assert(!pdu->is_retransmitted); + return pdu; failed: rpc_set_error(rpc, "zdr_callmsg failed with %s", @@ -375,6 +645,17 @@ void rpc_free_pdu(struct rpc_context *rpc, struct rpc_pdu *pdu) #endif /* HAVE_LIBKRB5 */ assert(rpc->magic == RPC_CONTEXT_MAGIC); + /* + * AZAUTH RPC is the only one queued with head priority and + * AZAUTH RPC MUST only be sent if use_azauth is true. + */ + assert(!pdu->is_head_prio || rpc->use_azauth); + +#ifdef ENABLE_PARANOID + /* PDU must be freed only after removing from all queues */ + assert(pdu->in_outqueue == PDU_ABSENT); + assert(pdu->in_waitpdu == PDU_ABSENT); +#endif if (pdu->zdr_decode_buf != NULL) { zdr_free(pdu->zdr_decode_fn, pdu->zdr_decode_buf); @@ -439,28 +720,61 @@ void pdu_set_timeout(struct rpc_context *rpc, struct rpc_pdu *pdu, uint64_t now_ #endif } + /* + * On major timeout we reset both major_timeout and timeout. + * Note that timeout can be updated multiple times before a major + * timeout, depending on the value of rpc->retrans. + */ if (pdu->major_timeout == 0) { pdu->major_timeout = now_msecs + (rpc->timeout * rpc->retrans); + pdu->timeout = now_msecs + rpc->timeout; #ifndef HAVE_CLOCK_GETTIME pdu->major_timeout += 1000; + pdu->timeout += 1000; #endif - /* Never less than pdu->timeout */ - if (pdu->major_timeout < pdu->timeout) { - pdu->major_timeout = pdu->timeout; - } + /* + * Early on when rpc->retrans is not set or if user doesn't + * set rpc->retrans, make sure major_timeout is set same as + * timeout. + */ + if (pdu->major_timeout < pdu->timeout) { + pdu->major_timeout = pdu->timeout; + } } } -int rpc_queue_pdu(struct rpc_context *rpc, struct rpc_pdu *pdu) +/** + * Queue pdu to rpc->outqueue. + * PDU is queued to the tail of outqueue unless high_prio is set, in which case + * it's queued to the head of outqueue (safe against partially sent head pdu). + * high_prio queueing may be useful for non-IO (non READ/WRITE) RPCs so they + * can be promptly sent out and they do not have to wait behind possibly huge + * number of WRITE/READ RPCs in the queue. Those (especially WRITE RPCs) can + * take very large time causing commands like stat/ls/find etc to appear to + * hang. + */ +int rpc_queue_pdu2(struct rpc_context *rpc, struct rpc_pdu *pdu, int prio) { int i, size = 0, pos; uint32_t recordmarker; + /* + * First pdu added to an empty outqueue is special, as an optimization + * we send it inline from here. Since there is no other pdu being sent + * it's safe against mixing bytes from different pdus. + * First high prio pdu is also sent inline even if there are other low + * priority pdus present in outqueue. + */ + bool_t send_now; #ifdef HAVE_LIBKRB5 uint32_t maj, min, val, len; gss_buffer_desc message_buffer, output_token; char *buf; #endif /* HAVE_LIBKRB5 */ + assert(prio == PDU_Q_PRIO_LOW || + prio == PDU_Q_PRIO_HI || + prio == PDU_Q_PRIO_HEAD); + assert(rpc->magic == RPC_CONTEXT_MAGIC); #ifdef HAVE_LIBKRB5 @@ -568,6 +882,9 @@ int rpc_queue_pdu(struct rpc_context *rpc, struct rpc_pdu *pdu) recordmarker = htonl(size | 0x80000000); memcpy(pdu->out.iov[0].buf, &recordmarker, 4); + /* 4 bytes for the recordmarker */ + pdu->req_size = size + 4; + /* * For udp we dont queue, we just send it straight away. * @@ -639,6 +956,15 @@ int rpc_queue_pdu(struct rpc_context *rpc, struct rpc_pdu *pdu) #endif /* HAVE_MULTITHREADING */ rpc_enqueue(&rpc->waitpdu[hash], pdu); rpc->waitpdu_len++; + +#ifdef ENABLE_PARANOID + assert(pdu->in_outqueue == PDU_ABSENT); + assert(pdu->in_waitpdu == PDU_ABSENT); + pdu->in_waitpdu = PDU_PRESENT; + pdu->added_to_waitpdu_at_line = __LINE__; + pdu->added_to_waitpdu_at_time = rpc_wallclock_time(); +#endif + #ifdef HAVE_MULTITHREADING if (rpc->multithreading_enabled) { nfs_mt_mutex_unlock(&rpc->rpc_mutex); @@ -653,19 +979,83 @@ int rpc_queue_pdu(struct rpc_context *rpc, struct rpc_pdu *pdu) nfs_mt_mutex_lock(&rpc->rpc_mutex); } #endif /* HAVE_MULTITHREADING */ - rpc_enqueue(&rpc->outqueue, pdu); + /* Fresh PDU being queued to outqueue, num_done must be 0 */ + assert(pdu->out.num_done == 0); + + if (prio == PDU_Q_PRIO_LOW) { + rpc_add_to_outqueue_lowp(rpc, pdu); + } else if (prio == PDU_Q_PRIO_HI) { + rpc_add_to_outqueue_highp(rpc, pdu); + } else { + rpc_add_to_outqueue_headp(rpc, pdu); + } + + send_now = (rpc->outqueue.head == pdu); + +#ifdef ENABLE_PARANOID + assert(pdu->in_waitpdu == PDU_ABSENT); + assert(pdu->in_outqueue == PDU_ABSENT); + pdu->in_outqueue = PDU_PRESENT; + pdu->added_to_outqueue_at_line = __LINE__; + pdu->added_to_outqueue_at_time = rpc_wallclock_time(); +#endif + #ifdef HAVE_MULTITHREADING if (rpc->multithreading_enabled) { nfs_mt_mutex_unlock(&rpc->rpc_mutex); } #endif /* HAVE_MULTITHREADING */ - if (rpc->outqueue.head == pdu) { + + /* + * If only PDU or a high/head priority PDU, send inline. + */ + if (send_now) { + /* + * We need to check if the token has expired, before we issue + * the RPC, else we can have the following problem: + * - user has not used the mount for a long time, and in the + * meantime the token expired. + * - now user uses the mount which issues a command from fuse. + * - the command comes here and since it's the first request + * to be queued in outqueue, send_now is true and we send the + * request over to the server. + * - server fails the requuest with "permission denied" as the + * auth token has expired. + * + * If the token has expired we do not send the request, but + * instead wake up rpc_service() thread, which again calls + * rpc_auth_needs_refresh() and triggers a reconnect. + * This will queue the AZAUTH RPC ahead of this request, + * perform the reconnect and auth refresh and once the refresh + * is successful, issue this new request. + */ + if (rpc_auth_needs_refresh(rpc)) { + RPC_LOG(rpc, 2, "Waking up rpc_service to refresh " + "auth token, not sending pdu %p", + pdu); + + /* + * Wakeup rpc_service() thread which will refresh the + * cert and issue the RPC after that. + */ + uint64_t evwrite = 1; + [[maybe_unused]] ssize_t evbytes = + write(rpc_get_evfd(rpc), &evwrite, sizeof(evwrite)); + assert(evbytes == 8); + return 0; + } + rpc_write_to_socket(rpc); } return 0; } +int rpc_queue_pdu(struct rpc_context *rpc, struct rpc_pdu *pdu) +{ + return rpc_queue_pdu2(rpc, pdu, PDU_Q_PRIO_LOW); +} + static int rpc_process_reply(struct rpc_context *rpc, ZDR *zdr) { struct rpc_msg msg; @@ -707,6 +1097,15 @@ static int rpc_process_reply(struct rpc_context *rpc, ZDR *zdr) pdu->cb(rpc, RPC_STATUS_ERROR, "RPC Packet not accepted by the server", pdu->private_data); return 0; } + + /* + * resp_size must be set to at least the size of the decoded headers. + * For READ RPCs it'll be more as it'll also include the data received. + * For zero-copy reads resp_size will be updated later as we read data + * bytes into the user zerop-copy buffer(s). + */ + assert(pdu->resp_size >= zdr->pos); + switch (msg.body.rbody.reply.areply.stat) { case SUCCESS: /* Last RPC response time for tracking RPC transport health */ @@ -983,6 +1382,12 @@ struct rpc_pdu *rpc_find_pdu(struct rpc_context *rpc, uint32_t xid) * but track previous entry for optimised removal */ prev_pdu = NULL; for (pdu=q->head; pdu; pdu=pdu->next) { + +#ifdef ENABLE_PARANOID + assert(pdu->in_outqueue == PDU_ABSENT); + assert(pdu->in_waitpdu == PDU_PRESENT); +#endif + if (pdu->xid != rpc->rm_xid[1]) { prev_pdu = pdu; continue; @@ -996,10 +1401,21 @@ struct rpc_pdu *rpc_find_pdu(struct rpc_context *rpc, uint32_t xid) if (prev_pdu != NULL) prev_pdu->next = pdu->next; rpc->waitpdu_len--; + +#ifdef ENABLE_PARANOID + pdu->in_waitpdu = PDU_ABSENT; + pdu->removed_from_waitpdu_at_line = __LINE__; + pdu->removed_from_waitpdu_at_time = rpc_wallclock_time(); +#endif + } break; } + if (pdu) { + pdu->next = NULL; + } + #ifdef HAVE_MULTITHREADING if (rpc->multithreading_enabled) { nfs_mt_mutex_unlock(&rpc->rpc_mutex); @@ -1009,6 +1425,30 @@ struct rpc_pdu *rpc_find_pdu(struct rpc_context *rpc, uint32_t xid) return pdu; } +bool_t rpc_pdu_is_retransmitted(struct rpc_pdu *pdu) +{ + return pdu->is_retransmitted; +} + +uint32_t rpc_pdu_get_req_size(struct rpc_pdu *pdu) +{ + return pdu->req_size; +} + +uint32_t rpc_pdu_get_resp_size(struct rpc_pdu *pdu) +{ + return pdu->resp_size; +} + +uint64_t rpc_pdu_get_dispatch_usecs(struct rpc_pdu *pdu) +{ +#ifdef HAVE_CLOCK_GETTIME + return pdu->dispatch_usecs; +#else + return 0; +#endif +} + int rpc_cancel_pdu(struct rpc_context *rpc, struct rpc_pdu *pdu) { /* @@ -1053,3 +1493,83 @@ int rpc_process_pdu(struct rpc_context *rpc, char *buf, int size) return 0; } +#ifdef ENABLE_PARANOID +/* + * Perfom extensive validation on the rpc_context and the various pdu queues. + * This helps to catch bugs related to pdu queues. + * rpc->rpc_mutex exclusive lock must be held by the caller. + */ +void rpc_paranoid_checks(struct rpc_context *rpc) +{ + struct rpc_pdu *pdu, *next_pdu, *last_pdu = NULL; + struct rpc_pdu *last_highprio_pdu = NULL; + int outqueue_count = 0; + int waitpdu_count = 0; + int i; + + for (pdu = rpc->outqueue.head; pdu; pdu = pdu->next) { + /* + * Must be present in outqueue and not waitpdu queue. + */ + assert(pdu->in_outqueue == PDU_PRESENT); + assert(pdu->in_waitpdu == PDU_ABSENT); + + /* + * Fully sent PDU should not be sitting in outqueue. + */ + assert(pdu->out.num_done < pdu->out.total_size); + + /* + * added_to_outqueue_at_time must be the latest. + */ + assert(pdu->added_to_outqueue_at_time > + pdu->removed_from_outqueue_at_time); + assert(pdu->added_to_outqueue_at_time > + pdu->added_to_waitpdu_at_time); + assert(pdu->added_to_outqueue_at_time >= + pdu->removed_from_waitpdu_at_time); + outqueue_count++; + last_pdu = pdu; + if (pdu->is_high_prio) + last_highprio_pdu = pdu; + } + assert(rpc->stats.outqueue_len == outqueue_count); + assert(rpc->outqueue.tail == last_pdu); + assert(rpc->outqueue.tailp == last_highprio_pdu); + + for (i = 0; i < rpc->num_hashes; i++) { + struct rpc_queue *q = &rpc->waitpdu[i]; + last_pdu = NULL; + for (pdu = q->head; pdu; pdu = next_pdu) { + next_pdu = pdu->next; + + /* + * Must be present in waitpdu queue and not outqueue. + */ + assert(pdu->in_waitpdu == PDU_PRESENT); + assert(pdu->in_outqueue == PDU_ABSENT); + + /* + * Only fully sent PDU should be sitting in waitpdu hash. + */ + assert(pdu->out.num_done == pdu->out.total_size); + + /* + * added_to_outqueue_at_time must be the latest. + */ + assert(pdu->added_to_waitpdu_at_time > + pdu->removed_from_waitpdu_at_time); + assert(pdu->added_to_waitpdu_at_time > + pdu->added_to_outqueue_at_time); + assert(pdu->added_to_waitpdu_at_time >= + pdu->removed_from_outqueue_at_time); + + waitpdu_count++; + last_pdu = pdu; + } + assert(q->tail == last_pdu); + } + + assert(rpc->waitpdu_len == waitpdu_count); +} +#endif diff --git a/lib/socket.c b/lib/socket.c index fcf2ab49..68f27513 100644 --- a/lib/socket.c +++ b/lib/socket.c @@ -107,6 +107,9 @@ static int rpc_reconnect_requeue(struct rpc_context *rpc); +static int +rpc_set_sockaddr(struct rpc_context *rpc, const char *server, int port); + static int create_socket(int domain, int type, int protocol) { @@ -141,7 +144,7 @@ set_nonblocking(int fd) static void set_nolinger(int fd) { -#if !defined(PS2_EE) +#if !defined(PS2_EE) struct linger lng; lng.l_onoff = 1; lng.l_linger = 0; @@ -257,10 +260,36 @@ rpc_get_fd(struct rpc_context *rpc) return rpc->fd; } -static int -rpc_has_queue(struct rpc_queue *q) +int +rpc_get_evfd(struct rpc_context *rpc) { - return q->head != NULL; + return rpc->evfd; +} + +/* + * Does rpc->outqueue have one or more PDUs waiting to be sent out. + */ +static bool_t +rpc_outqueue_present(struct rpc_context *rpc) +{ + bool_t present; + +#ifdef HAVE_MULTITHREADING + if (rpc->multithreading_enabled) { + nfs_mt_mutex_lock(&rpc->rpc_mutex); + } +#endif /* HAVE_MULTITHREADING */ + + present = (rpc->outqueue.head != NULL); + assert(present == (rpc->stats.outqueue_len != 0)); + +#ifdef HAVE_MULTITHREADING + if (rpc->multithreading_enabled) { + nfs_mt_mutex_unlock(&rpc->rpc_mutex); + } +#endif /* HAVE_MULTITHREADING */ + + return present; } int @@ -277,19 +306,10 @@ rpc_which_events(struct rpc_context *rpc) return POLLIN; } -#ifdef HAVE_MULTITHREADING - if (rpc->multithreading_enabled) { - nfs_mt_mutex_lock(&rpc->rpc_mutex); - } -#endif /* HAVE_MULTITHREADING */ - if (rpc_has_queue(&rpc->outqueue)) { - events |= POLLOUT; - } -#ifdef HAVE_MULTITHREADING - if (rpc->multithreading_enabled) { - nfs_mt_mutex_unlock(&rpc->rpc_mutex); + if (rpc_outqueue_present(rpc)) { + events |= POLLOUT; } -#endif /* HAVE_MULTITHREADING */ + return events; } @@ -315,15 +335,21 @@ rpc_write_to_socket(struct rpc_context *rpc) } #endif /* HAVE_MULTITHREADING */ +#ifdef ENABLE_PARANOID + rpc_paranoid_checks(rpc); +#endif + /* Write several pdus at once */ while ((rpc->max_waitpdu_len == 0 || rpc->max_waitpdu_len > rpc->waitpdu_len) && (pdu = rpc->outqueue.head) != NULL) { + int niov = 0; uint32_t num_pdus = 0; char *last_buf = NULL; ssize_t count; + assert(rpc->stats.outqueue_len > 0); assert(pdu->out.niov <= pdu->out.iov_capacity); assert(pdu->out.iov_capacity <= RPC_MAX_VECTORS); @@ -345,10 +371,51 @@ rpc_write_to_socket(struct rpc_context *rpc) } do { + /* + * AZAUTH RPC is the only one queued with head priority and + * AZAUTH RPC MUST only be sent if use_azauth is true. + */ + assert(!pdu->is_head_prio || rpc->use_azauth); + + /* + * If context needs auth and connection is not authorized (yet), + * only ever send AZAUTH RPCs out. + */ + if (rpc->use_azauth && + !rpc->auth_context.is_authorized && + !pdu->is_head_prio) { + RPC_LOG(rpc, 2, "Not sending queued RPC pdu %p as " + "connection is not authorized", pdu); + /* + * If we have something to write, write it, else + * exit. Note that the only pdu we will be + * writing would be the AZAUTH RPC. + */ + if (niov) { + break; + } else { + ret = 0; + goto finished; + } + } + size_t num_done = pdu->out.num_done; int pdu_niov = pdu->out.niov; int i; +#ifdef ENABLE_PARANOID + assert(pdu->in_outqueue == PDU_PRESENT); + assert(pdu->in_waitpdu == PDU_ABSENT); + + if (pdu->is_head_prio) { + RPC_LOG(rpc, 2, "rpc_write_to_socket: Sending " + "AZAUTH PDU %p", pdu); + } + +#endif + /* Fully sent PDU should not be sitting in outqueue */ + assert(num_done < pdu->out.total_size); + for (i = 0; i < pdu_niov; i++) { char *buf = pdu->out.iov[i].buf; size_t len = pdu->out.iov[i].len; @@ -380,20 +447,47 @@ rpc_write_to_socket(struct rpc_context *rpc) rpc->max_waitpdu_len > (rpc->waitpdu_len + num_pdus)) && pdu != NULL && niov < iovcnt); + /* + * We must never be doing 0-byte writes as those can get into + * infinite loop. + */ + assert(niov > 0); + count = writev(rpc->fd, iov, niov); if (count == -1) { if (errno == EAGAIN || errno == EWOULDBLOCK) { - ret = 0; - goto finished; + /* + * Update EAGAIN stats to get an idea about the + * receive window advertised by the peer. + * Once write() returns EAGAIN, we only come back + * to write after poll() returns POLLOUT, which + * it'll do only when there's enough sndbuf space + * in the socket. + * + * XXX This assert has been seen to fail sometimes. + * Maybe Windows TCP is reneg'ing the window. + * Anyways, disabling the assert. + */ +#if 0 + assert(rpc->stats.last_write_bytes_before_eagain > 0); +#endif + rpc->stats.tot_write_bytes_before_eagain += + rpc->stats.last_write_bytes_before_eagain; + rpc->stats.last_write_bytes_before_eagain = 0; + INC_STATS(rpc, num_write_eagain); + ret = 0; + goto finished; } rpc_set_error_locked(rpc, "Error when writing to " - "socket :%d %s", errno, - rpc_get_error(rpc)); + "socket :%d %s", errno, + rpc_get_error(rpc)); ret = -1; goto finished; } + rpc->stats.last_write_bytes_before_eagain += count; + /* Check how many pdu we completed */ while (count > 0 && (pdu = rpc->outqueue.head) != NULL) { size_t remaining = (pdu->out.total_size - pdu->out.num_done); @@ -405,12 +499,36 @@ rpc_write_to_socket(struct rpc_context *rpc) pdu->out.num_done = pdu->out.total_size; rpc->outqueue.head = pdu->next; - if (pdu->next == NULL) + if (rpc->outqueue.head == NULL) rpc->outqueue.tail = NULL; + /* + * Last high priority pdu dequeued, no more + * high priority pdus in outqueue. + */ + if (rpc->outqueue.tailp == pdu) + rpc->outqueue.tailp = NULL; + + assert(rpc->stats.outqueue_len > 0); + rpc->stats.outqueue_len--; + +#ifdef ENABLE_PARANOID + pdu->in_outqueue = PDU_ABSENT; + pdu->removed_from_outqueue_at_line = __LINE__; + pdu->removed_from_outqueue_at_time = rpc_wallclock_time(); +#endif + /* RPC sent, original or retransmit */ INC_STATS(rpc, num_req_sent); +#ifdef HAVE_CLOCK_GETTIME + /* + * Now this RPC is completely written over the socket. + * Note current wallclock time as the dispatch time. + */ + pdu->dispatch_usecs = rpc_wallclock_time(); +#endif + if (pdu->flags & PDU_DISCARD_AFTER_SENDING) { rpc_free_pdu(rpc, pdu); ret = 0; @@ -420,14 +538,25 @@ rpc_write_to_socket(struct rpc_context *rpc) hash = rpc_hash_xid(rpc, pdu->xid); rpc_enqueue(&rpc->waitpdu[hash], pdu); rpc->waitpdu_len++; +#ifdef ENABLE_PARANOID + pdu->in_waitpdu = PDU_PRESENT; + pdu->added_to_waitpdu_at_line = __LINE__; + pdu->added_to_waitpdu_at_time = rpc_wallclock_time(); +#endif + } else { pdu->out.num_done += count; + assert(pdu->out.num_done < pdu->out.total_size); break; } } } finished: +#ifdef ENABLE_PARANOID + rpc_paranoid_checks(rpc); +#endif + #ifdef HAVE_MULTITHREADING if (rpc->multithreading_enabled) { nfs_mt_mutex_unlock(&rpc->rpc_mutex); @@ -512,7 +641,7 @@ static void rpc_finished_pdu(struct rpc_context *rpc) } rpc->state = READ_RM; rpc->inpos = 0; - if (rpc->is_udp == 0 || rpc->is_broadcast == 0) { + if (rpc->pdu && (rpc->is_udp == 0 || rpc->is_broadcast == 0)) { rpc_free_pdu(rpc, rpc->pdu); rpc->pdu = NULL; } @@ -559,6 +688,12 @@ rpc_read_from_socket(struct rpc_context *rpc) free(buf); return -1; } + + /* + * For UDP, the entire RPC PDU is received at once. + */ + rpc->pdu->resp_size = count; + if (rpc_process_pdu(rpc, buf, count) != 0) { rpc_set_error(rpc, "Invalid/garbage pdu received from " "server. Ignoring PDU"); @@ -681,12 +816,22 @@ rpc_read_from_socket(struct rpc_context *rpc) } rpc->inpos += count; + /* + * As we read RPC PDU data, update the response size in + * pdu->resp_size. + * Caller can query this using rpc_pdu_get_resp_size() inside + * the callback. + */ + if (rpc->pdu) { + rpc->pdu->resp_size += count; + } + if (rpc->buf) { rpc->buf += count; } else { rpc_advance_cursor(rpc, &rpc->pdu->in, count); } - + if (rpc->inpos == rpc->pdu_size) { switch (rpc->state) { case READ_RM: @@ -745,7 +890,15 @@ rpc_read_from_socket(struct rpc_context *rpc) * here will force a reconnect, which will anyways * re-queue everything from waitpdu[] to outqueue. */ - rpc_return_to_queue(&rpc->outqueue, rpc->pdu); +#ifdef ENABLE_PARANOID + assert(rpc->pdu->in_waitpdu == PDU_ABSENT); + assert(rpc->pdu->in_outqueue == PDU_ABSENT); + rpc->pdu->in_outqueue = PDU_PRESENT; + rpc->pdu->added_to_outqueue_at_line = __LINE__; + rpc->pdu->added_to_outqueue_at_time = rpc_wallclock_time(); +#endif + rpc_return_to_outqueue(rpc, rpc->pdu); + rpc->pdu = NULL; #ifdef HAVE_MULTITHREADING if (rpc->multithreading_enabled) { @@ -767,7 +920,7 @@ rpc_read_from_socket(struct rpc_context *rpc) * that we have already read these 4 bytes in * PAYLOAD and FRAGMENT */ - rpc->inpos = 0; + rpc->inpos = 0; if (!rpc->is_server_context) { /* Unknown xid, either unsolicited @@ -777,6 +930,11 @@ rpc_read_from_socket(struct rpc_context *rpc) rpc->state = READ_UNKNOWN; continue; } + + /* + * RM + XID. + */ + rpc->pdu->resp_size = 8; } continue; case READ_FRAGMENT: @@ -818,6 +976,15 @@ rpc_read_from_socket(struct rpc_context *rpc) #endif /* HAVE_LIBKRB5 */ /* We do not have rpc->pdu for server context */ if (rpc->pdu && rpc->pdu->free_zdr) { + /* + * If the READ failed, bail out here as there is no + * data. + */ + const READ3res *res = (READ3res *) rpc->pdu->zdr_decode_buf; + if (res->status != NFS3_OK) { + goto payload_finished; + } + /* * We are doing zero-copy read. * pdu->read_count is the amount of read data returned by @@ -873,9 +1040,7 @@ rpc_read_from_socket(struct rpc_context *rpc) continue; } } -#ifdef HAVE_LIBKRB5 payload_finished: -#endif /* HAVE_LIBKRB5 */ if (rpc->fragments) { free(rpc->buf); rpc->buf = NULL; @@ -947,6 +1112,13 @@ rpc_timeout_scan(struct rpc_context *rpc) nfs_mt_mutex_lock(&rpc->rpc_mutex); } #endif /* HAVE_MULTITHREADING */ + + /* + * First check requests that have timed out while sitting in outqueue. + * These have not been sent to the server so do not indicate any issue + * with server or connection, hence we do not take any corrective + * action based on these request timeouts. + */ for (pdu = rpc->outqueue.head; pdu; pdu = next_pdu) { next_pdu = pdu->next; @@ -960,47 +1132,58 @@ rpc_timeout_scan(struct rpc_context *rpc) } /* Timed out w/o being sent */ - INC_STATS(rpc, num_timedout); + INC_STATS(rpc, num_timedout_in_outqueue); + + /* + * rpc->retrans > 0 implies that user wants us to retransmit + * timed out RPCs. Note that we treat non-zero rpc->retrans + * as hard mount, so we just advance the timeout values for + * this RPC and leave it in the outqueue. + * Since these have not been sent to the server, they don't + * signify any issue with the server or the connection and + * hence major timeout has no special significance for such + * requests. + */ + if (!pdu->do_not_retry && rpc->retrans > 0) { + /* + * Ask pdu_set_timeout() to advance pdu->timeout and + * pdu->major_timeout. Note that major_timeout has no + * special significance for requests timing out in + * outqueue. + */ + pdu->timeout = 0; + pdu->major_timeout = 0; + pdu_set_timeout(rpc, pdu, t); + + RPC_LOG(rpc, 2, "[pdu %p] Request timed out in outqueue, " + "will send when connection allows!", pdu); + } else { + rpc_remove_pdu_from_queue(&rpc->outqueue, pdu); + assert(rpc->stats.outqueue_len > 0); + rpc->stats.outqueue_len--; + +#ifdef ENABLE_PARANOID + assert(pdu->in_outqueue == PDU_PRESENT); + assert(pdu->in_waitpdu == PDU_ABSENT); + pdu->in_outqueue = PDU_ABSENT; + pdu->removed_from_outqueue_at_line = __LINE__; + pdu->removed_from_outqueue_at_time = rpc_wallclock_time(); +#endif - /* - * rpc->retrans > 0 implies that user wants us to retransmit - * timed out RPCs. We update the timeout values for this RPC - * and leave it in the outqueue. - */ - if (!pdu->do_not_retry && rpc->retrans > 0) { - /* Ask pdu_set_timeout() to set pdu->timeout */ - pdu->timeout = 0; - - if (t >= pdu->major_timeout) { - /* Timed out w/o being sent */ - INC_STATS(rpc, num_major_timedout); - - /* Ask pdu_set_timeout() to set pdu->major_timeout */ - pdu->major_timeout = 0; - if (!pdu->snr_logged) { - /* Log only once for an RPC */ - pdu->snr_logged = TRUE; - RPC_LOG(rpc, 1, "[pdu %p] Server %s not " - "responding, still trying", - pdu, rpc->server); - } - if (!need_reconnect) { - need_reconnect = (last_rpc_msecs > rpc->timeout); - } - } - /* Reset the RPC timeout values as appropriate */ - pdu_set_timeout(rpc, pdu, t); - } else { - LIBNFS_LIST_REMOVE(&rpc->outqueue.head, pdu); - if (!rpc->outqueue.head) { - rpc->outqueue.tail = NULL; //done - } rpc_set_error_locked(rpc, "command timed out"); pdu->cb(rpc, RPC_STATUS_TIMEOUT, NULL, pdu->private_data); rpc_free_pdu(rpc, pdu); } } + + /* + * Now look for requests in waitpdu. These are requests which have + * been sent to server and we are awaiting response from the server. + * These may indicate an unresponsive server and/or bad connection. + * We log a message on major_timeout and try recovery by dropping + * existing connection and creting a new one. + */ for (i = 0; i < rpc->num_hashes; i++) { struct rpc_queue *q; @@ -1012,24 +1195,32 @@ rpc_timeout_scan(struct rpc_context *rpc) /* no timeout for this pdu */ continue; } - if (t < pdu->timeout) { + if (t < pdu->timeout && t < pdu->major_timeout) { /* not expired yet */ continue; } /* Timed out waiting for response */ - INC_STATS(rpc, num_timedout); + if (t >= pdu->timeout) { + INC_STATS(rpc, num_timedout); + } - LIBNFS_LIST_REMOVE(&q->head, pdu); - if (!q->head) { - q->tail = NULL; - } + rpc_remove_pdu_from_queue(q, pdu); rpc->waitpdu_len--; +#ifdef ENABLE_PARANOID + assert(pdu->in_outqueue == PDU_ABSENT); + assert(pdu->in_waitpdu == PDU_PRESENT); + pdu->in_waitpdu = PDU_ABSENT; + pdu->removed_from_waitpdu_at_line = __LINE__; + pdu->removed_from_waitpdu_at_time = rpc_wallclock_time(); +#endif + /* - * rpc->retrans > 0 implies that user wants us to retransmit - * timed out RPCs. We update the timeout values for this RPC - * and move it to the outqueue. + * rpc->retrans > 0 implies that user wants us to + * retransmit timed out RPCs. We update the timeout + * values for these RPCs and move them to outqueue for + * retransmit. */ if (!pdu->do_not_retry && rpc->retrans > 0) { /* Ask pdu_set_timeout() to set pdu->timeout */ @@ -1056,13 +1247,16 @@ rpc_timeout_scan(struct rpc_context *rpc) pdu_set_timeout(rpc, pdu, t); /* queue it back to outqueue for retransmit */ - rpc_return_to_queue(&rpc->outqueue, pdu); - - /* Retransmit on timeout */ - INC_STATS(rpc, num_retransmitted); + rpc_return_to_outqueue(rpc, pdu); + +#ifdef ENABLE_PARANOID + assert(pdu->in_waitpdu == PDU_ABSENT); + assert(pdu->in_outqueue == PDU_ABSENT); + pdu->in_outqueue = PDU_PRESENT; + pdu->added_to_outqueue_at_line = __LINE__; + pdu->added_to_outqueue_at_time = rpc_wallclock_time(); +#endif - /* we have to re-send the whole pdu again */ - pdu->out.num_done = 0; } else { // qqq move to a temporary queue and process after // we drop the mutex @@ -1087,6 +1281,71 @@ rpc_timeout_scan(struct rpc_context *rpc) return (need_reconnect ? -1 : 0); } + +/* + * Returns TRUE when the auth is enabled for the connection and token has + * expired, needing a refresh. + * We need to reconnect the connection in this case, to refresh the token. + * This is edge trigerred, i.e., it returns true till we initiate the + * reconnect (which will eventually refresh the token) and not till the token + * is refreshed. + */ +bool_t +rpc_auth_needs_refresh(struct rpc_context *rpc) +{ + /* + * If not using azauth, we should not proceed further and return from + * here. + */ + if (!rpc->use_azauth) { + return FALSE; + } + + /* + * Once marked "needs refresh" return true till it's reset. + * This must be placed before the is_authorized check below, as on + * expiry we set is_authorized to FALSE and needs_refresh to TRUE. + */ + if (rpc->auth_context.needs_refresh) { + return TRUE; + } + + /* + * If connection is not authorized, we need not check for expiry time + * as it will not be set. + * It is important to check because rpc_service will be running and this + * function will keep getting called. We should check for expiry only when + * connection is authorized. + */ + if (!rpc->auth_context.is_authorized) { + return FALSE; + } + + /* + * Refresh token sufficiently before expiry, to avoid situation where + * we send some RPC request(s) to the server and by the time they are + * processed at the server, token expires and the requests are failed. + * 5 min should be sufficient, as no request can sit in the server for + * more than ~1 min. + */ + const uint64_t refresh_at = rpc->auth_context.expiry_time - 300; + const uint64_t now = (uint64_t) time(NULL); + + assert((int64_t) refresh_at > 0); + + if (rpc->auth_context.is_authorized && now >= refresh_at) { + RPC_LOG(rpc, 1, "Auth token about to expire (or expired), " + "reconnecting to acquire a new token. " + "refresh_at: %ld, now: %ld", + refresh_at, now); + rpc->auth_context.is_authorized = FALSE; + rpc->auth_context.needs_refresh = TRUE; + return TRUE; + } + + return FALSE; +} + int rpc_service(struct rpc_context *rpc, int revents) { @@ -1098,7 +1357,7 @@ rpc_service(struct rpc_context *rpc, int revents) * connection. Schedule reconnect and requeue and return. Once the new * connection is ready, events will be processed for that. */ - if (rpc_timeout_scan(rpc) != 0) { + if ((rpc_timeout_scan(rpc) != 0) || rpc_auth_needs_refresh(rpc)) { return rpc_reconnect_requeue(rpc); } @@ -1232,21 +1491,21 @@ rpc_service(struct rpc_context *rpc, int revents) } } + if ((revents & POLLOUT) && rpc_outqueue_present(rpc)) { #ifdef HAVE_TLS - /* - * For secure NFS connections we should never write to the socket w/o - * properly completing the TLS handshake. Note that we do allow reads - * from the socket as we would want to read response to the AUTH_TLS - * NULL RPC. - */ - if (rpc->use_tls && (rpc->tls_context.state != TLS_HANDSHAKE_COMPLETED)) { - RPC_LOG(rpc, 2, "TLS handshake state %d on fd %d, skipping socket write!", - rpc->tls_context.state, rpc->fd); - return 0; - } + /* + * For secure NFS connections we should never write to the socket w/o + * properly completing the TLS handshake. Note that we do allow reads + * from the socket as we would want to read response to the AUTH_TLS + * NULL RPC. + */ + if (rpc->use_tls && (rpc->tls_context.state != TLS_HANDSHAKE_COMPLETED)) { + RPC_LOG(rpc, 2, "TLS handshake state %d on fd %d, skipping socket write!", + rpc->tls_context.state, rpc->fd); + return 0; + } #endif - if (revents & POLLOUT && rpc_has_queue(&rpc->outqueue)) { if (rpc_write_to_socket(rpc) != 0) { if (rpc->is_server_context) { return -1; @@ -1318,34 +1577,64 @@ rpc_connect_sockaddr_async(struct rpc_context *rpc) assert(rpc->magic == RPC_CONTEXT_MAGIC); + /* + * If the user has configured "resolve on reconnect" behaviour, then + * we must resolve server name fresh and not connect to the last + * resolved address. + */ + if (rpc->resolve_server) { + /* Can only be set if user opted for it */ + assert(rpc->resolve_on_reconnect); + + /* + * resolve_server must be set only on reconnect, which means + * address must have been resolved earlier, and hence port + * must be set. + */ + assert(rpc->port != 0); + assert(rpc->server != NULL); + + RPC_LOG(rpc, 2, "Resolving server %s on reconnect (port %d)", + rpc->server, rpc->port); + + if (rpc_set_sockaddr(rpc, rpc->server, rpc->port) != 0) { + return -1; + } + rpc->resolve_server = 0; + } + switch (s->ss_family) { case AF_INET: socksize = sizeof(struct sockaddr_in); rpc->fd = create_socket(AF_INET, SOCK_STREAM, IPPROTO_TCP); - if (set_bind_device(rpc->fd, rpc->ifname) != 0) { - rpc_set_error (rpc, "Failed to bind to interface"); - return -1; - } + if (rpc->fd != -1) { + if (set_bind_device(rpc->fd, rpc->ifname) != 0) { + rpc_set_error (rpc, "Failed to bind to interface"); + return -1; + } #ifdef HAVE_NETINET_TCP_H - if (rpc->tcp_syncnt != RPC_PARAM_UNDEFINED) { - set_tcp_sockopt(rpc->fd, TCP_SYNCNT, rpc->tcp_syncnt); - } + if (rpc->tcp_syncnt != RPC_PARAM_UNDEFINED) { + set_tcp_sockopt(rpc->fd, TCP_SYNCNT, rpc->tcp_syncnt); + } #endif + } break; case AF_INET6: socksize = sizeof(struct sockaddr_in6); rpc->fd = create_socket(AF_INET6, SOCK_STREAM, IPPROTO_TCP); - if (set_bind_device(rpc->fd, rpc->ifname) != 0) { - rpc_set_error (rpc, "Failed to bind to interface"); - return -1; - } + if (rpc->fd != -1) { + if (set_bind_device(rpc->fd, rpc->ifname) != 0) { + rpc_set_error (rpc, "Failed to bind to interface"); + return -1; + } #ifdef HAVE_NETINET_TCP_H - if (rpc->tcp_syncnt != RPC_PARAM_UNDEFINED) { - set_tcp_sockopt(rpc->fd, TCP_SYNCNT, rpc->tcp_syncnt); - } + if (rpc->tcp_syncnt != RPC_PARAM_UNDEFINED) { + set_tcp_sockopt(rpc->fd, TCP_SYNCNT, rpc->tcp_syncnt); + } #endif + } break; default: rpc_set_error(rpc, "Can not handle AF_FAMILY:%d", s->ss_family); @@ -1353,7 +1642,7 @@ rpc_connect_sockaddr_async(struct rpc_context *rpc) } if (rpc->fd == -1) { - rpc_set_error(rpc, "Failed to open socket"); + rpc_set_error(rpc, "Failed to open socket: %s", strerror(errno)); return -1; } @@ -1398,7 +1687,7 @@ rpc_connect_sockaddr_async(struct rpc_context *rpc) { struct sockaddr_storage ss; struct sockaddr_in *sin; -#if !defined(PS3_PPU) && !defined(PS2_EE) +#if !defined(PS3_PPU) && !defined(PS2_EE) struct sockaddr_in6 *sin6; #endif static int portOfs = 0; @@ -1407,7 +1696,7 @@ rpc_connect_sockaddr_async(struct rpc_context *rpc) int startOfs, port, rc; sin = (struct sockaddr_in *)&ss; -#if !defined(PS3_PPU) && !defined(PS2_EE) +#if !defined(PS3_PPU) && !defined(PS2_EE) sin6 = (struct sockaddr_in6 *)&ss; #endif if (portOfs == 0) { @@ -1484,15 +1773,76 @@ static int rpc_set_sockaddr(struct rpc_context *rpc, const char *server, int port) { struct addrinfo *ai = NULL; + int err, i; - if (getaddrinfo(server, NULL, NULL, &ai) != 0) { - rpc_set_error(rpc, "Invalid address:%s. " - "Can not resolv into IPv4/v6 structure.", server); - return -1; - } + assert(strcmp(rpc->server, server) == 0); + /* + * When called on reconnect (rpc->resolve_server is set) rpc->port must + * be set and must match the port passed. + */ + assert(!rpc->resolve_server || (rpc->port == port)); + + for (i = 0; i < 100; i++) { + if ((err = getaddrinfo(server, NULL, NULL, &ai)) != 0) { + if (err == EAI_AGAIN) { + RPC_LOG(rpc, 2, "rpc_set_sockaddr: getaddrinfo(%s) failed " + "temporarily, trying again!", server); + usleep(1000 * 100); + continue; + } + + rpc_set_error(rpc, "Invalid address:%s. " + "Can not resolv into IPv4/v6 structure: %s (%s)", + server, gai_strerror(err), strerror(errno)); + return -1; + } + + err = 0; + break; + } + + if (err != 0) { + rpc_set_error(rpc, "Failed to resolve address %s, even after 100 retries!", + server); + return -1; + } switch (ai->ai_family) { case AF_INET: + /* + * If not the first call to rpc_set_sockaddr(), address family + * must match what's already saved. Port can be different as + * the same rpc_context may connect to portmapper, mount and + * nfs, all over different ports. Address can change in case + * of migration of NFS server. + */ + assert(((struct sockaddr_in *)&rpc->s)->sin_family == AF_UNSPEC || + ((struct sockaddr_in *)&rpc->s)->sin_family == AF_INET); + + if (rpc->resolve_server) { + char ip_str_old[INET_ADDRSTRLEN]; + char ip_str_new[INET_ADDRSTRLEN]; + + inet_ntop(AF_INET, &(((struct sockaddr_in *)&rpc->s)->sin_addr), + ip_str_old, INET_ADDRSTRLEN); + inet_ntop(AF_INET, &(((struct sockaddr_in *)(ai->ai_addr))->sin_addr), + ip_str_new, INET_ADDRSTRLEN); + + assert(((struct sockaddr_in *)&rpc->s)->sin_family == AF_INET); + assert(((struct sockaddr_in *)&rpc->s)->sin_addr.s_addr != 0); + assert(((struct sockaddr_in *)&rpc->s)->sin_port == htons(port)); + + if (((struct sockaddr_in *)&rpc->s)->sin_addr.s_addr != + ((struct sockaddr_in *)(ai->ai_addr))->sin_addr.s_addr) { + RPC_LOG(rpc, 1, "rpc_set_sockaddr (%s): IPv4 address changed " + "from %s -> %s", server, + ip_str_old, ip_str_new); + } else { + RPC_LOG(rpc, 2, "rpc_set_sockaddr (%s): IPv4 address again " + "resolved to %s", server, ip_str_new); + } + } + ((struct sockaddr_in *)&rpc->s)->sin_family = ai->ai_family; ((struct sockaddr_in *)&rpc->s)->sin_port = htons(port); ((struct sockaddr_in *)&rpc->s)->sin_addr = @@ -1504,6 +1854,9 @@ rpc_set_sockaddr(struct rpc_context *rpc, const char *server, int port) break; #if !defined(PS3_PPU) && !defined(PS2_EE) case AF_INET6: + assert(((struct sockaddr_in *)&rpc->s)->sin_family == AF_UNSPEC || + ((struct sockaddr_in *)&rpc->s)->sin_family == AF_INET6); + ((struct sockaddr_in6 *)&rpc->s)->sin6_family = ai->ai_family; ((struct sockaddr_in6 *)&rpc->s)->sin6_port = htons(port); ((struct sockaddr_in6 *)&rpc->s)->sin6_addr = @@ -1517,6 +1870,11 @@ rpc_set_sockaddr(struct rpc_context *rpc, const char *server, int port) } freeaddrinfo(ai); + /* + * Note the port for reconnect. + */ + rpc->port = port; + return 0; } @@ -1586,6 +1944,43 @@ rpc_disconnect(struct rpc_context *rpc, const char *error) } #ifdef HAVE_TLS + +/* + * During TCP reconnection, for secure transport, we need to re-perform auth. + * This is the callback function called when auth completes. +*/ +static void +reconnect_cb_azauth(struct rpc_context *rpc, int status, + void *command_data, void *private_data) +{ + /* reconnect_cb_tls() passes NULL as private_data */ + assert(private_data == NULL); + + assert(rpc->magic == RPC_CONTEXT_MAGIC); + + /* Must be called only for TLS transport */ + assert(rpc->use_azauth); + + /* + * During reconnect, if azauth fails, we have no choice but to keep + * trying. + */ + if (!rpc->auth_context.is_authorized) { + RPC_LOG(rpc, 1, "reconnect_cb_azauth: AZAUTH failed, " + "restarting connection!"); + + if (rpc->fd != -1) { + close(rpc->fd); + rpc->fd = -1; + } + rpc->is_connected = 0; + rpc_reconnect_requeue(rpc); + return; + } + + RPC_LOG(rpc, 2, "reconnect_cb_azauth: AzAuth completed successfully!"); +} + /* * During TCP reconnection (either server or client closes connection) for secure * transport we need to perform the TLS handshake. This is the callback function @@ -1595,6 +1990,8 @@ static void reconnect_cb_tls(struct rpc_context *rpc, int status, void *command_data, void *private_data) { + /* reconnect_cb() passes NULL as private_data */ + assert(private_data == NULL); assert(rpc->magic == RPC_CONTEXT_MAGIC); /* Must be called only for TLS transport */ @@ -1621,8 +2018,27 @@ reconnect_cb_tls(struct rpc_context *rpc, int status, } RPC_LOG(rpc, 2, "reconnect_cb_tls: TLS handshake completed successfully!"); + + /* + * TLS handshake completed successfully. + * If azauth is enabled, perform it now. + */ + if (rpc->use_azauth) { + RPC_LOG(rpc, 2, "reconnect_cb_tls: sending AZAUTH RPC"); + + if (rpc_perform_azauth(rpc, reconnect_cb_azauth, NULL) == NULL) { + RPC_LOG(rpc, 1, "reconnect_cb_azauth: rpc_perform_azauth() failed, " + "restarting connection!"); + if (rpc->fd != -1) { + close(rpc->fd); + rpc->fd = -1; + } + rpc->is_connected = 0; + rpc_reconnect_requeue(rpc); + } + } } -#endif +#endif /* HAVE_TLS */ static void reconnect_cb(struct rpc_context *rpc, int status, void *data, @@ -1630,6 +2046,8 @@ reconnect_cb(struct rpc_context *rpc, int status, void *data, { assert(rpc->magic == RPC_CONTEXT_MAGIC); + RPC_LOG(rpc, 2, "reconnect_cb called with status %d", status); + if (status != RPC_STATUS_SUCCESS) { rpc_set_error(rpc, "Failed to reconnect async"); rpc_reconnect_requeue(rpc); @@ -1644,8 +2062,9 @@ reconnect_cb(struct rpc_context *rpc, int status, void *data, /* * For secure NFS connections, we need to setup TLS session now. */ - RPC_LOG(rpc, 2, "reconnect_cb called with status %d", status); if (rpc->use_tls) { + RPC_LOG(rpc, 2, "reconnect_cb: sending AUTH_TLS"); + if (rpc_null_task_authtls(rpc, rpc->nfs_version, reconnect_cb_tls, NULL) == NULL) { RPC_LOG(rpc, 1, "reconnect_cb: rpc_null_task_authtls() failed, " @@ -1667,6 +2086,29 @@ reconnect_cb(struct rpc_context *rpc, int status, void *data, } } #endif /* HAVE_TLS */ + +#ifdef ENABLE_INSECURE_AUTH_FOR_DEVTEST + else if (rpc->use_azauth) { + /* + * Insecure connection, if azauth is enabled perform auth. + * + * Note: THIS WOULD SEND THE TOKEN OVER AN INSECURE CONNECTION + * AND MUST ONLY BE USED IN DEVTEST ON TRUSTED NETWORKS. + */ + RPC_LOG(rpc, 2, "reconnect_cb: sending insecure AZAUTH RPC"); + + if (rpc_perform_azauth(rpc, reconnect_cb_azauth, NULL) == NULL) { + RPC_LOG(rpc, 1, "reconnect_cb: rpc_perform_azauth() failed, " + "restarting connection!"); + if (rpc->fd != -1) { + close(rpc->fd); + rpc->fd = -1; + } + rpc->is_connected = 0; + rpc_reconnect_requeue(rpc); + } + } +#endif } /* Disconnect but do not error all PDUs, just move pdus in-flight back to the @@ -1695,14 +2137,17 @@ rpc_reconnect_requeue(struct rpc_context *rpc) } rpc->fd = -1; rpc->is_connected = 0; - - if (rpc->outqueue.head) { - rpc->outqueue.head->out.num_done = 0; - } - rpc->inpos = 0; rpc->state = READ_RM; + /* + * As part of reconnect handling, auth token will be refreshed if + * needed, now we can clear needs_refresh. Note the reconnect handling + * is resilient and it'll keep trying till reconnect (and anything else + * needed, i.e., TLS handshake, and/or auth refresh) succeeds. + */ + rpc->auth_context.needs_refresh = FALSE; + /* Socket is closed so we will not get any replies to any commands * in flight. Move them all over from the waitpdu queue back to the * out queue. @@ -1712,19 +2157,80 @@ rpc_reconnect_requeue(struct rpc_context *rpc) nfs_mt_mutex_lock(&rpc->rpc_mutex); } #endif /* HAVE_MULTITHREADING */ + + if (rpc->outqueue.head) { + rpc->outqueue.head->out.num_done = 0; + + /* + * If there's an AZAUTH RPC in the outqueue, remove it as a + * fresh AZAUTH RPC is issued on reconnect. + */ + pdu = rpc->outqueue.head; + + if (pdu->is_head_prio) { + RPC_LOG(rpc, 1, "rpc_reconnect_requeue: Removing AZAUTH " + "RPC pdu %p from outqueue", pdu); + + rpc->outqueue.head = pdu->next; + if (rpc->outqueue.head == NULL) + rpc->outqueue.tail = NULL; + + /* + * Last high priority pdu dequeued, no more + * high priority pdus in outqueue. + */ + if (rpc->outqueue.tailp == pdu) + rpc->outqueue.tailp = NULL; + + assert(rpc->stats.outqueue_len > 0); + rpc->stats.outqueue_len--; + +#ifdef ENABLE_PARANOID + pdu->in_outqueue = PDU_ABSENT; + pdu->removed_from_outqueue_at_line = __LINE__; + pdu->removed_from_outqueue_at_time = rpc_wallclock_time(); +#endif + + rpc_free_pdu(rpc, pdu); + } + } + for (i = 0; i < rpc->num_hashes; i++) { struct rpc_queue *q = &rpc->waitpdu[i]; for (pdu = q->head; pdu; pdu = next) { next = pdu->next; - rpc_return_to_queue(&rpc->outqueue, pdu); - /* Retransmit on reconnect */ - INC_STATS(rpc, num_retransmitted); - /* we have to re-send the whole pdu again */ - pdu->out.num_done = 0; + rpc_return_to_outqueue(rpc, pdu); + +#ifdef ENABLE_PARANOID + assert(pdu->in_waitpdu == PDU_PRESENT); + assert(pdu->in_outqueue == PDU_ABSENT); + pdu->in_waitpdu = PDU_ABSENT; + pdu->in_outqueue = PDU_PRESENT; + pdu->removed_from_waitpdu_at_line = __LINE__; + pdu->removed_from_waitpdu_at_time = rpc_wallclock_time(); + pdu->added_to_outqueue_at_line = __LINE__; + pdu->added_to_outqueue_at_time = rpc_wallclock_time(); +#endif } rpc_reset_queue(q); } rpc->waitpdu_len = 0; + + /* + * If there's any half-read PDU, that needs to be restarted too. + */ + if (rpc->pdu) { +#ifdef ENABLE_PARANOID + assert(rpc->pdu->in_waitpdu == PDU_ABSENT); + assert(rpc->pdu->in_outqueue == PDU_ABSENT); + rpc->pdu->in_outqueue = PDU_PRESENT; + rpc->pdu->added_to_outqueue_at_line = __LINE__; + rpc->pdu->added_to_outqueue_at_time = rpc_wallclock_time(); +#endif + rpc_return_to_outqueue(rpc, rpc->pdu); + rpc->pdu = NULL; + } + #ifdef HAVE_MULTITHREADING if (rpc->multithreading_enabled) { nfs_mt_mutex_unlock(&rpc->rpc_mutex); @@ -1734,12 +2240,20 @@ rpc_reconnect_requeue(struct rpc_context *rpc) if (rpc->auto_reconnect < 0 || rpc->num_retries > 0) { rpc->num_retries--; rpc->connect_cb = reconnect_cb; - RPC_LOG(rpc, 1, "reconnect initiated"); + RPC_LOG(rpc, 1, "reconnect initiated to %s", rpc->server); + /* + * If user has opted for "resolve on reconnect", let + * rpc_connect_sockaddr_async() know that. + */ + rpc->resolve_server = rpc->resolve_on_reconnect; if (rpc_connect_sockaddr_async(rpc) != 0) { rpc_error_all_pdus(rpc, "RPC ERROR: Failed to " "reconnect async"); return -1; } + /* rpc_connect_sockaddr_async() must have reset it */ + assert(!rpc->resolve_server); + INC_STATS(rpc, num_reconnects); return 0; } @@ -1850,20 +2364,28 @@ int rpc_queue_length(struct rpc_context *rpc) { int i = 0; +#ifdef ENABLE_PARANOID struct rpc_pdu *pdu; +#endif assert(rpc->magic == RPC_CONTEXT_MAGIC); - for(pdu = rpc->outqueue.head; pdu; pdu = pdu->next) { - i++; - } - #ifdef HAVE_MULTITHREADING if (rpc->multithreading_enabled) { nfs_mt_mutex_lock(&rpc->rpc_mutex); } #endif /* HAVE_MULTITHREADING */ + +#ifdef ENABLE_PARANOID + for(pdu = rpc->outqueue.head; pdu; pdu = pdu->next) { + i++; + } + assert(rpc->stats.outqueue_len == i); +#else + i = rpc->stats.outqueue_len; +#endif i += rpc->waitpdu_len; + #ifdef HAVE_MULTITHREADING if (rpc->multithreading_enabled) { nfs_mt_mutex_unlock(&rpc->rpc_mutex); @@ -1893,6 +2415,18 @@ rpc_set_fd(struct rpc_context *rpc, int fd) rpc->fd = fd; } +void +rpc_set_resolve_on_reconnect(struct rpc_context *rpc) +{ + assert(rpc->magic == RPC_CONTEXT_MAGIC); + /* + * It's not a sin to call rpc_set_resolve_on_reconnect() more than + * once but our callers shouldn't call so catch that. + */ + assert(!rpc->resolve_on_reconnect); + rpc->resolve_on_reconnect = 1; +} + int rpc_is_udp_socket(struct rpc_context *rpc) { diff --git a/nfs/libnfs-raw-nfs.c b/nfs/libnfs-raw-nfs.c index f7cbb68b..6fea172a 100644 --- a/nfs/libnfs-raw-nfs.c +++ b/nfs/libnfs-raw-nfs.c @@ -425,6 +425,48 @@ zdr_COMMIT3res (ZDR *zdrs, COMMIT3res *objp) return TRUE; } +uint32_t +zdr_AZAUTH3args (ZDR *zdrs, AZAUTH3args *objp) +{ + if (!zdr_string (zdrs, &objp->client_version, 16)) + return FALSE; + if (!zdr_string (zdrs, &objp->clientid, 64)) + return FALSE; + if (!zdr_string (zdrs, &objp->authtype, 16)) + return FALSE; + if (!zdr_string (zdrs, &objp->authtarget, 256)) + return FALSE; + if (!zdr_string (zdrs, &objp->authdata, 16384)) + return FALSE; + return TRUE; +} + +uint32_t +zdr_AZAUTH3resok (ZDR *zdrs, AZAUTH3resok *objp) +{ + if (!zdr_string (zdrs, &objp->server_version, 16)) + return FALSE; + if (!zdr_string (zdrs, &objp->serverid, 64)) + return FALSE; + return TRUE; +} + +uint32_t +zdr_AZAUTH3res (ZDR *zdrs, AZAUTH3res *objp) +{ + if (!zdr_nfsstat3 (zdrs, &objp->status)) + return FALSE; + switch (objp->status) { + case NFS3_OK: + if (!zdr_AZAUTH3resok (zdrs, &objp->AZAUTH3res_u.resok)) + return FALSE; + break; + default: + break; + } + return TRUE; +} + uint32_t zdr_ACCESS3args (ZDR *zdrs, ACCESS3args *objp) { diff --git a/nfs/libnfs-raw-nfs.h b/nfs/libnfs-raw-nfs.h index 05a6fc4c..c695833f 100644 --- a/nfs/libnfs-raw-nfs.h +++ b/nfs/libnfs-raw-nfs.h @@ -281,6 +281,29 @@ struct COMMIT3res { } COMMIT3res_u; }; typedef struct COMMIT3res COMMIT3res; + +struct AZAUTH3args { + char *client_version; + char *clientid; + char *authtype; + char *authtarget; + char *authdata; +}; +typedef struct AZAUTH3args AZAUTH3args; + +struct AZAUTH3resok { + char *server_version; + char *serverid; +}; +typedef struct AZAUTH3resok AZAUTH3resok; + +struct AZAUTH3res { + nfsstat3 status; + union { + AZAUTH3resok resok; + } AZAUTH3res_u; +}; +typedef struct AZAUTH3res AZAUTH3res; #define ACCESS3_READ 0x0001 #define ACCESS3_LOOKUP 0x0002 #define ACCESS3_MODIFY 0x0004 @@ -1527,6 +1550,9 @@ extern PATHCONF3res * nfs3_pathconf_3_svc(PATHCONF3args *, struct svc_req *); #define NFS3_COMMIT 21 extern COMMIT3res * nfs3_commit_3(COMMIT3args *, void *); extern COMMIT3res * nfs3_commit_3_svc(COMMIT3args *, struct svc_req *); +#define NFS3_AZAUTH 23 +extern AZAUTH3res * nfs3_azauth_3(AZAUTH3args *, void *); +extern AZAUTH3res * nfs3_azauth_3_svc(AZAUTH3args *, struct svc_req *); extern int nfs_program_3_freeresult (void *, zdrproc_t, caddr_t); #else /* K&R C */ @@ -1596,6 +1622,9 @@ extern PATHCONF3res * nfs3_pathconf_3_svc(); #define NFS3_COMMIT 21 extern COMMIT3res * nfs3_commit_3(); extern COMMIT3res * nfs3_commit_3_svc(); +#define NFS3_AZAUTH 23 +extern AZAUTH3res * nfs3_azauth_3(); +extern AZAUTH3res * nfs3_azauth_3_svc(); extern int nfs_program_3_freeresult (); #endif /* K&R C */ @@ -1665,6 +1694,9 @@ extern uint32_t zdr_COMMIT3args (ZDR *, COMMIT3args*); extern uint32_t zdr_COMMIT3resok (ZDR *, COMMIT3resok*); extern uint32_t zdr_COMMIT3resfail (ZDR *, COMMIT3resfail*); extern uint32_t zdr_COMMIT3res (ZDR *, COMMIT3res*); +extern uint32_t zdr_AZAUTH3args (ZDR *, AZAUTH3args*); +extern uint32_t zdr_AZAUTH3resok (ZDR *, AZAUTH3resok*); +extern uint32_t zdr_AZAUTH3res (ZDR *, AZAUTH3res*); extern uint32_t zdr_ACCESS3args (ZDR *, ACCESS3args*); extern uint32_t zdr_ACCESS3resok (ZDR *, ACCESS3resok*); extern uint32_t zdr_ACCESS3resfail (ZDR *, ACCESS3resfail*); @@ -1852,6 +1884,9 @@ extern uint32_t zdr_COMMIT3args (); extern uint32_t zdr_COMMIT3resok (); extern uint32_t zdr_COMMIT3resfail (); extern uint32_t zdr_COMMIT3res (); +extern uint32_t zdr_AZAUTH3args (); +extern uint32_t zdr_AZAUTH3resok (); +extern uint32_t zdr_AZAUTH3res (); extern uint32_t zdr_ACCESS3args (); extern uint32_t zdr_ACCESS3resok (); extern uint32_t zdr_ACCESS3resfail (); diff --git a/nfs/nfs.c b/nfs/nfs.c index 22f05251..0e1e5138 100644 --- a/nfs/nfs.c +++ b/nfs/nfs.c @@ -123,7 +123,7 @@ struct rpc_pdu *rpc_nfs3_null_task(struct rpc_context *rpc, rpc_cb cb, return NULL; } - if (rpc_queue_pdu(rpc, pdu) != 0) { + if (rpc_queue_pdu2(rpc, pdu, PDU_Q_PRIO_HI) != 0) { rpc_set_error(rpc, "Out of memory. Failed to queue pdu for NFS3/NULL call"); return NULL; } @@ -149,7 +149,7 @@ struct rpc_pdu *rpc_nfs3_getattr_task(struct rpc_context *rpc, rpc_cb cb, return NULL; } - if (rpc_queue_pdu(rpc, pdu) != 0) { + if (rpc_queue_pdu2(rpc, pdu, PDU_Q_PRIO_HI) != 0) { rpc_set_error(rpc, "Out of memory. Failed to queue pdu for NFS3/GETATTR call"); return NULL; } @@ -175,7 +175,7 @@ struct rpc_pdu *rpc_nfs3_pathconf_task(struct rpc_context *rpc, rpc_cb cb, return NULL; } - if (rpc_queue_pdu(rpc, pdu) != 0) { + if (rpc_queue_pdu2(rpc, pdu, PDU_Q_PRIO_HI) != 0) { rpc_set_error(rpc, "Out of memory. Failed to queue pdu for NFS3/PATHCONF call"); return NULL; } @@ -201,7 +201,7 @@ struct rpc_pdu *rpc_nfs3_lookup_task(struct rpc_context *rpc, rpc_cb cb, return NULL; } - if (rpc_queue_pdu(rpc, pdu) != 0) { + if (rpc_queue_pdu2(rpc, pdu, PDU_Q_PRIO_HI) != 0) { rpc_set_error(rpc, "Out of memory. Failed to queue pdu for NFS3/LOOKUP call"); return NULL; } @@ -227,7 +227,7 @@ struct rpc_pdu *rpc_nfs3_access_task(struct rpc_context *rpc, rpc_cb cb, return NULL; } - if (rpc_queue_pdu(rpc, pdu) != 0) { + if (rpc_queue_pdu2(rpc, pdu, PDU_Q_PRIO_HI) != 0) { rpc_set_error(rpc, "Out of memory. Failed to queue pdu for NFS3/ACCESS call"); return NULL; } @@ -309,7 +309,15 @@ rpc_nfs3_readv_task(struct rpc_context *rpc, rpc_cb cb, return NULL; } - pdu->in.base = (struct iovec *) malloc(sizeof(struct iovec) * iovcnt); + /* + * Allocate twice the iovec space, first half will be used for iov[]. + * This will be updated as data is read into user buffers. + * Second half is for iov_ref[]. This is not used in happy path. Only + * if we need to resend the request we need it to reset the cursor to + * the original iovec. + * See rpc_reset_cursor(). + */ + pdu->in.base = (struct iovec *) malloc(sizeof(struct iovec) * iovcnt * 2); if (!pdu->in.base) { rpc_set_error(rpc, "error: Failed to allocate memory"); rpc_free_pdu(rpc, pdu); @@ -317,16 +325,24 @@ rpc_nfs3_readv_task(struct rpc_context *rpc, rpc_cb cb, } pdu->in.iov = pdu->in.base; - pdu->in.iovcnt = iovcnt; + pdu->in.iov_ref = pdu->in.base + iovcnt; + pdu->in.iovcnt = pdu->in.iovcnt_ref = iovcnt; for (i = 0; i < iovcnt; i++) { - pdu->in.iov[i] = iov[i]; + pdu->in.iov[i] = pdu->in.iov_ref[i] = iov[i]; pdu->in.remaining_size += iov[i].iov_len; } pdu->requested_read_count = pdu->in.remaining_size; - if (rpc_queue_pdu(rpc, pdu) != 0) { + /* + * Add read requests to high priority queue so that they can be + * dispatched ahead of writes which can be large requests and too + * many writes queued can unnecessarily delay reads. By dispatching + * reads faster we can have them executed on the server and save + * undue delays. + */ + if (rpc_queue_pdu2(rpc, pdu, PDU_Q_PRIO_HI) != 0) { rpc_set_error(rpc, "Out of memory. Failed to queue pdu for NFS3/READ call"); return NULL; } @@ -400,14 +416,15 @@ struct rpc_pdu *rpc_nfs3_writev_task(struct rpc_context *rpc, rpc_cb cb, } /* - * We add 4 to the user provided iovcnt to account for one each for + * We add 5 to the user provided iovcnt to account for one each for * the following: * - Record marker * - RPC header * - NFS header + * - Write length * - Padding (optional) */ - pdu = rpc_allocate_pdu2(rpc, NFS_PROGRAM, NFS_V3, NFS3_WRITE, cb, private_data, (zdrproc_t)zdr_WRITE3res, sizeof(WRITE3res), 0, iovcnt + 4); + pdu = rpc_allocate_pdu2(rpc, NFS_PROGRAM, NFS_V3, NFS3_WRITE, cb, private_data, (zdrproc_t)zdr_WRITE3res, sizeof(WRITE3res), 0, iovcnt + 5); if (pdu == NULL) { rpc_set_error(rpc, "Out of memory. Failed to allocate pdu for NFS3/WRITE call"); return NULL; @@ -517,6 +534,37 @@ struct rpc_pdu *rpc_nfs3_commit_task(struct rpc_context *rpc, rpc_cb cb, return pdu; } + +struct rpc_pdu *rpc_nfs3_azauth_task(struct rpc_context *rpc, rpc_cb cb, + struct AZAUTH3args *args, + void *private_data) +{ + struct rpc_pdu *pdu; + + pdu = rpc_allocate_pdu(rpc, NFS_PROGRAM, NFS_V3, NFS3_AZAUTH, cb, private_data, (zdrproc_t)zdr_AZAUTH3res, sizeof(AZAUTH3res)); + if (pdu == NULL) { + rpc_set_error(rpc, "Out of memory. Failed to allocate pdu for NFS3/AZAUTH call"); + return NULL; + } + + if (zdr_AZAUTH3args(&pdu->zdr, args) == 0) { + rpc_set_error(rpc, "ZDR error: Failed to encode AZAUTH3args"); + rpc_free_pdu(rpc, pdu); + return NULL; + } + + /* + * We add AzAuth RPC to outqueue head as we want to send it before any + * other request(s). + */ + if (rpc_queue_pdu2(rpc, pdu, PDU_Q_PRIO_HEAD) != 0) { + rpc_set_error(rpc, "Out of memory. Failed to queue pdu for NFS3/AZAUTH call"); + return NULL; + } + + return pdu; +} + struct rpc_pdu * rpc_nfs3_setattr_task(struct rpc_context *rpc, rpc_cb cb, SETATTR3args *args, void *private_data) @@ -535,7 +583,7 @@ rpc_nfs3_setattr_task(struct rpc_context *rpc, rpc_cb cb, SETATTR3args *args, return NULL; } - if (rpc_queue_pdu(rpc, pdu) != 0) { + if (rpc_queue_pdu2(rpc, pdu, PDU_Q_PRIO_HI) != 0) { rpc_set_error(rpc, "Out of memory. Failed to queue pdu for NFS3/SETATTR call"); return NULL; } @@ -560,7 +608,7 @@ struct rpc_pdu *rpc_nfs3_mkdir_task(struct rpc_context *rpc, rpc_cb cb, return NULL; } - if (rpc_queue_pdu(rpc, pdu) != 0) { + if (rpc_queue_pdu2(rpc, pdu, PDU_Q_PRIO_HI) != 0) { rpc_set_error(rpc, "Out of memory. Failed to queue pdu for NFS3/MKDIR call"); return NULL; } @@ -586,7 +634,7 @@ struct rpc_pdu *rpc_nfs3_rmdir_task(struct rpc_context *rpc, rpc_cb cb, return NULL; } - if (rpc_queue_pdu(rpc, pdu) != 0) { + if (rpc_queue_pdu2(rpc, pdu, PDU_Q_PRIO_HI) != 0) { rpc_set_error(rpc, "Out of memory. Failed to queue pdu for NFS3/RMDIR call"); return NULL; } @@ -611,7 +659,7 @@ struct rpc_pdu *rpc_nfs3_create_task(struct rpc_context *rpc, rpc_cb cb, return NULL; } - if (rpc_queue_pdu(rpc, pdu) != 0) { + if (rpc_queue_pdu2(rpc, pdu, PDU_Q_PRIO_HI) != 0) { rpc_set_error(rpc, "Out of memory. Failed to queue pdu for NFS3/CREATE call"); return NULL; } @@ -637,7 +685,7 @@ struct rpc_pdu *rpc_nfs3_mknod_task(struct rpc_context *rpc, rpc_cb cb, return NULL; } - if (rpc_queue_pdu(rpc, pdu) != 0) { + if (rpc_queue_pdu2(rpc, pdu, PDU_Q_PRIO_HI) != 0) { rpc_set_error(rpc, "Out of memory. Failed to queue pdu for NFS3/MKNOD call"); return NULL; } @@ -663,7 +711,7 @@ struct rpc_pdu *rpc_nfs3_remove_task(struct rpc_context *rpc, rpc_cb cb, return NULL; } - if (rpc_queue_pdu(rpc, pdu) != 0) { + if (rpc_queue_pdu2(rpc, pdu, PDU_Q_PRIO_HI) != 0) { rpc_set_error(rpc, "Out of memory. Failed to queue pdu for NFS3/REMOVE call"); return NULL; } @@ -689,7 +737,7 @@ struct rpc_pdu *rpc_nfs3_readdir_task(struct rpc_context *rpc, rpc_cb cb, return NULL; } - if (rpc_queue_pdu(rpc, pdu) != 0) { + if (rpc_queue_pdu2(rpc, pdu, PDU_Q_PRIO_HI) != 0) { rpc_set_error(rpc, "Out of memory. Failed to queue pdu for NFS3/READDIR call"); return NULL; } @@ -715,7 +763,7 @@ struct rpc_pdu *rpc_nfs3_readdirplus_task(struct rpc_context *rpc, rpc_cb cb, return NULL; } - if (rpc_queue_pdu(rpc, pdu) != 0) { + if (rpc_queue_pdu2(rpc, pdu, PDU_Q_PRIO_HI) != 0) { rpc_set_error(rpc, "Out of memory. Failed to queue pdu for NFS3/READDIRPLUS call"); return NULL; } @@ -741,7 +789,7 @@ struct rpc_pdu *rpc_nfs3_fsstat_task(struct rpc_context *rpc, rpc_cb cb, return NULL; } - if (rpc_queue_pdu(rpc, pdu) != 0) { + if (rpc_queue_pdu2(rpc, pdu, PDU_Q_PRIO_HI) != 0) { rpc_set_error(rpc, "Out of memory. Failed to queue pdu for NFS3/FSSTAT call"); return NULL; } @@ -767,7 +815,7 @@ struct rpc_pdu *rpc_nfs3_fsinfo_task(struct rpc_context *rpc, rpc_cb cb, return NULL; } - if (rpc_queue_pdu(rpc, pdu) != 0) { + if (rpc_queue_pdu2(rpc, pdu, PDU_Q_PRIO_HI) != 0) { rpc_set_error(rpc, "Out of memory. Failed to queue pdu for NFS3/FSINFO call"); return NULL; } @@ -793,7 +841,7 @@ rpc_nfs3_readlink_task(struct rpc_context *rpc, rpc_cb cb, return NULL; } - if (rpc_queue_pdu(rpc, pdu) != 0) { + if (rpc_queue_pdu2(rpc, pdu, PDU_Q_PRIO_HI) != 0) { rpc_set_error(rpc, "Out of memory. Failed to queue pdu for NFS3/READLINK call"); return NULL; } @@ -806,7 +854,12 @@ struct rpc_pdu *rpc_nfs3_symlink_task(struct rpc_context *rpc, rpc_cb cb, { struct rpc_pdu *pdu; - pdu = rpc_allocate_pdu(rpc, NFS_PROGRAM, NFS_V3, NFS3_SYMLINK, cb, private_data, (zdrproc_t)zdr_SYMLINK3res, sizeof(SYMLINK3res)); + /* + * symlink target max length is 4096 bytes and we need space to encode + * symlink attributes too, so we use additional 512 bytes for request + * encoding. + */ + pdu = rpc_allocate_pdu2(rpc, NFS_PROGRAM, NFS_V3, NFS3_SYMLINK, cb, private_data, (zdrproc_t)zdr_SYMLINK3res, sizeof(SYMLINK3res), 512, 0); if (pdu == NULL) { rpc_set_error(rpc, "Out of memory. Failed to allocate pdu for NFS3/SYMLINK call"); return NULL; @@ -818,7 +871,7 @@ struct rpc_pdu *rpc_nfs3_symlink_task(struct rpc_context *rpc, rpc_cb cb, return NULL; } - if (rpc_queue_pdu(rpc, pdu) != 0) { + if (rpc_queue_pdu2(rpc, pdu, PDU_Q_PRIO_HI) != 0) { rpc_set_error(rpc, "Out of memory. Failed to queue pdu for NFS3/SYMLINK call"); return NULL; } @@ -844,7 +897,7 @@ struct rpc_pdu *rpc_nfs3_rename_task(struct rpc_context *rpc, rpc_cb cb, return NULL; } - if (rpc_queue_pdu(rpc, pdu) != 0) { + if (rpc_queue_pdu2(rpc, pdu, PDU_Q_PRIO_HI) != 0) { rpc_set_error(rpc, "Out of memory. Failed to queue pdu for NFS3/RENAME call"); return NULL; } @@ -869,7 +922,7 @@ struct rpc_pdu *rpc_nfs3_link_task(struct rpc_context *rpc, rpc_cb cb, return NULL; } - if (rpc_queue_pdu(rpc, pdu) != 0) { + if (rpc_queue_pdu2(rpc, pdu, PDU_Q_PRIO_HI) != 0) { rpc_set_error(rpc, "Out of memory. Failed to queue pdu for NFS3/LINK call"); return NULL; } diff --git a/nfs/nfs.x b/nfs/nfs.x index 5f7afe57..4653ad63 100644 --- a/nfs/nfs.x +++ b/nfs/nfs.x @@ -237,6 +237,26 @@ union COMMIT3res switch (nfsstat3 status) { COMMIT3resfail resfail; }; +struct AZAUTH3args { + string client_version<16>; + string clientid<64>; + string authtype<16>; + string authtarget<256>; + string authdata<16384>; +}; + +struct AZAUTH3resok { + string server_version<16>; + string serverid<64>; +}; + +union AZAUTH3res switch (nfsstat3 status) { + case NFS3_OK: + AZAUTH3resok resok; + default: + void; +}; + const ACCESS3_READ = 0x0001; const ACCESS3_LOOKUP = 0x0002; const ACCESS3_MODIFY = 0x0004; @@ -1196,6 +1216,9 @@ program NFS_PROGRAM { COMMIT3res NFS3_COMMIT(COMMIT3args) = 21; + + AZAUTH3res + NFS3_AZAUTH(AZAUTH3args) = 23; } = 3; } = 100003; diff --git a/nfs4/nfs4.c b/nfs4/nfs4.c index 47a9557a..d4f3391b 100644 --- a/nfs4/nfs4.c +++ b/nfs4/nfs4.c @@ -294,18 +294,19 @@ struct rpc_pdu *rpc_nfs4_readv_task(struct rpc_context *rpc, rpc_cb cb, return NULL; } - pdu->in.base = (struct iovec *) malloc(sizeof(struct iovec) * iovcnt); - if (!pdu->in.base) { - rpc_set_error(rpc, "error: Failed to allocate memory"); - rpc_free_pdu(rpc, pdu); - return NULL; - } + pdu->in.base = (struct iovec *) malloc(sizeof(struct iovec) * iovcnt * 2); + if (!pdu->in.base) { + rpc_set_error(rpc, "error: Failed to allocate memory"); + rpc_free_pdu(rpc, pdu); + return NULL; + } pdu->in.iov = pdu->in.base; - pdu->in.iovcnt = iovcnt; + pdu->in.iov_ref = pdu->in.base + iovcnt; + pdu->in.iovcnt = pdu->in.iovcnt_ref = iovcnt; for (i = 0; i < iovcnt; i++) { - pdu->in.iov[i] = iov[i]; + pdu->in.iov[i] = pdu->in.iov_ref[i] = iov[i]; pdu->in.remaining_size += iov[i].iov_len; }