Skip to content

Commit

Permalink
Loading
Loading status checks…
Fixes formatting after rebase
Browse files Browse the repository at this point in the history
@ilumsden
ilumsden committed Dec 16, 2024
1 parent c0ea2b2 commit fc7ba50
Showing 6 changed files with 1,224 additions and 1,201 deletions.
26 changes: 12 additions & 14 deletions include/dyad/client/dyad_client.h
View file
Original file line number Diff line number Diff line change
@@ -28,14 +28,14 @@
extern "C" {
#endif
#if DYAD_PERFFLOW
#define DYAD_CORE_FUNC_MODS __attribute__((annotate("@critical_path()"))) static
#define DYAD_CORE_FUNC_MODS __attribute__ ((annotate ("@critical_path()"))) static
#else
#define DYAD_CORE_FUNC_MODS static inline
#endif

struct dyad_metadata {
char *fpath;
uint32_t owner_rank;
char *fpath;
uint32_t owner_rank;
};
typedef struct dyad_metadata dyad_metadata_t;

@@ -47,8 +47,7 @@ typedef struct dyad_metadata dyad_metadata_t;
*
* @return An error code from dyad_rc.h
*/
DYAD_PFA_ANNOTATE DYAD_DLL_EXPORTED dyad_rc_t dyad_produce(dyad_ctx_t *ctx,
const char *fname);
DYAD_PFA_ANNOTATE DYAD_DLL_EXPORTED dyad_rc_t dyad_produce (dyad_ctx_t *ctx, const char *fname);

/**
* @brief Obtain DYAD metadata for a file in the consumer-managed directory
@@ -61,12 +60,12 @@ DYAD_PFA_ANNOTATE DYAD_DLL_EXPORTED dyad_rc_t dyad_produce(dyad_ctx_t *ctx,
*
* @return An error code from dyad_rc.h
*/
DYAD_PFA_ANNOTATE DYAD_DLL_EXPORTED dyad_rc_t
dyad_get_metadata(dyad_ctx_t *ctx, const char *fname, bool should_wait,
dyad_metadata_t **mdata);
DYAD_PFA_ANNOTATE DYAD_DLL_EXPORTED dyad_rc_t dyad_get_metadata (dyad_ctx_t *ctx,
const char *fname,
bool should_wait,
dyad_metadata_t **mdata);

DYAD_PFA_ANNOTATE DYAD_DLL_EXPORTED dyad_rc_t
dyad_free_metadata(dyad_metadata_t **mdata);
DYAD_PFA_ANNOTATE DYAD_DLL_EXPORTED dyad_rc_t dyad_free_metadata (dyad_metadata_t **mdata);

/**
* @brief Wrapper function that performs all the common tasks needed
@@ -76,8 +75,7 @@ dyad_free_metadata(dyad_metadata_t **mdata);
*
* @return An error code from dyad_rc.h
*/
DYAD_PFA_ANNOTATE DYAD_DLL_EXPORTED dyad_rc_t dyad_consume(dyad_ctx_t *ctx,
const char *fname);
DYAD_PFA_ANNOTATE DYAD_DLL_EXPORTED dyad_rc_t dyad_consume (dyad_ctx_t *ctx, const char *fname);

/**
* @brief Wrapper function that performs all the common tasks needed
@@ -89,8 +87,8 @@ DYAD_PFA_ANNOTATE DYAD_DLL_EXPORTED dyad_rc_t dyad_consume(dyad_ctx_t *ctx,
*
* @return An error code from dyad_rc.h
*/
DYAD_PFA_ANNOTATE DYAD_DLL_EXPORTED dyad_rc_t dyad_consume_w_metadata(
dyad_ctx_t *ctx, const char *fname, const dyad_metadata_t *mdata);
DYAD_PFA_ANNOTATE DYAD_DLL_EXPORTED dyad_rc_t
dyad_consume_w_metadata (dyad_ctx_t *ctx, const char *fname, const dyad_metadata_t *mdata);

#ifdef __cplusplus
}
2 changes: 1 addition & 1 deletion include/dyad/common/dyad_structures.h
View file
Original file line number Diff line number Diff line change
@@ -10,4 +10,4 @@
struct dyad_ctx;
typedef struct dyad_ctx dyad_ctx_t;

#endif /* DYAD_COMMON_STRUCTURES_H */
#endif /* DYAD_COMMON_STRUCTURES_H */
1,689 changes: 855 additions & 834 deletions src/dyad/client/dyad_client.c
View file

Large diffs are not rendered by default.

64 changes: 33 additions & 31 deletions src/dyad/client/dyad_client_int.h
View file
Original file line number Diff line number Diff line change
@@ -9,53 +9,55 @@
// Debug message
#ifndef DPRINTF
#if VA_OPT_SUPPORTED
#define DPRINTF(curr_dyad_ctx, fmt, ...) \
do { \
if ((curr_dyad_ctx) && (curr_dyad_ctx)->debug) \
fprintf(stderr, (fmt)__VA_OPT__(, ) __VA_ARGS__); \
} while (0)
#define DPRINTF(curr_dyad_ctx, fmt, ...) \
do { \
if ((curr_dyad_ctx) && (curr_dyad_ctx)->debug) \
fprintf (stderr, (fmt)__VA_OPT__ (, ) __VA_ARGS__); \
} while (0)
#else
#define DPRINTF(curr_dyad_ctx, fmt, ...) \
do { \
if ((curr_dyad_ctx) && (curr_dyad_ctx)->debug) \
fprintf(stderr, (fmt), ##__VA_ARGS__); \
} while (0)
#define DPRINTF(curr_dyad_ctx, fmt, ...) \
do { \
if ((curr_dyad_ctx) && (curr_dyad_ctx)->debug) \
fprintf (stderr, (fmt), ##__VA_ARGS__); \
} while (0)
#endif
#endif // DPRINTF
#endif // DPRINTF

#define TIME_DIFF(Tstart, Tend) \
((double)(1000000000L * ((Tend).tv_sec - (Tstart).tv_sec) + (Tend).tv_nsec - \
(Tstart).tv_nsec) / \
1000000000L)
#define TIME_DIFF(Tstart, Tend) \
((double)(1000000000L * ((Tend).tv_sec - (Tstart).tv_sec) + (Tend).tv_nsec - (Tstart).tv_nsec) \
/ 1000000000L)

// Detailed information message that can be omitted
#if DYAD_FULL_DEBUG
#define IPRINTF DPRINTF
#define IPRINTF_DEFINED
#else
#define IPRINTF(curr_dyad_ctx, fmt, ...)
#endif // DYAD_FULL_DEBUG
#endif // DYAD_FULL_DEBUG

DYAD_DLL_EXPORTED int gen_path_key(const char *str, char *path_key,
const size_t len, const uint32_t depth,
const uint32_t width);
DYAD_DLL_EXPORTED int gen_path_key (const char *str,
char *path_key,
const size_t len,
const uint32_t depth,
const uint32_t width);

/**
* Private Function definitions
*/
DYAD_DLL_EXPORTED dyad_rc_t dyad_get_data(const dyad_ctx_t *ctx,
const dyad_metadata_t *mdata,
char **file_data, size_t *file_len);
DYAD_DLL_EXPORTED dyad_rc_t dyad_commit(dyad_ctx_t *ctx, const char *fname);

DYAD_DLL_EXPORTED dyad_rc_t dyad_kvs_read(const dyad_ctx_t *ctx,
const char *topic, const char *upath,
bool should_wait,
dyad_metadata_t **mdata);
DYAD_DLL_EXPORTED dyad_rc_t dyad_get_data (const dyad_ctx_t *ctx,
const dyad_metadata_t *mdata,
char **file_data,
size_t *file_len);
DYAD_DLL_EXPORTED dyad_rc_t dyad_commit (dyad_ctx_t *ctx, const char *fname);

DYAD_DLL_EXPORTED dyad_rc_t dyad_kvs_read (const dyad_ctx_t *ctx,
const char *topic,
const char *upath,
bool should_wait,
dyad_metadata_t **mdata);

#if DYAD_SYNC_DIR
DYAD_PFA_ANNOTATE DYAD_DLL_EXPORTED int dyad_sync_directory(dyad_ctx_t *ctx,
const char *path);
DYAD_PFA_ANNOTATE DYAD_DLL_EXPORTED int dyad_sync_directory (dyad_ctx_t *ctx, const char *path);
#endif

#endif /* DYAD_CORE_DYAD_CORE_INT_H */
#endif /* DYAD_CORE_DYAD_CORE_INT_H */
71 changes: 35 additions & 36 deletions src/dyad/common/dyad_structures_int.h
View file
Original file line number Diff line number Diff line change
@@ -24,46 +24,45 @@ extern "C" {
* @struct dyad_ctx
*/
struct dyad_ctx {
// Internal
void *h; // the Flux handle for DYAD
struct dyad_dtl *dtl_handle; // Opaque handle to DTL info
const char *fname; // Used to track which file is getting processed.
bool use_fs_locks; // Used to track if fs locks should be used.
char *prod_real_path; // producer managed real path
char *cons_real_path; // consumer managed real path
uint32_t prod_managed_len; // length of producer path managed by DYAD
uint32_t cons_managed_len; // length of consumer path managed by DYAD
uint32_t prod_real_len; // length of producer managed real path
uint32_t cons_real_len; // length of consumer managed real path
uint32_t prod_managed_hash; // hash of producer path managed by DYAD
uint32_t cons_managed_hash; // hash of consumer path managed by DYAD
uint32_t prod_real_hash; // hash of producer managed real path
uint32_t cons_real_hash; // hash of consumer managed real path
uint32_t delim_len; // length of path delimiter
// User Facing
bool debug; // if true, perform debug logging
bool check; // if true, perform some check logging
bool reenter; // if false, do not recursively enter DYAD
bool initialized; // if true, DYAD is initialized
bool shared_storage; // if true, the managed path is shared
bool async_publish; // Enable asynchronous publish by producer
bool fsync_write; // Apply fsync after write by producer
unsigned int key_depth; // Depth of bins for the Flux KVS
unsigned int key_bins; // Number of bins for the Flux KVS
uint32_t rank; // Flux rank for DYAD
uint32_t service_mux; // Number of Flux brokers sharing node-local storage
uint32_t node_idx; // Index of the node hosting broker(s)
int pid; // unix process id, obtained by getpid()
char *kvs_namespace; // Flux KVS namespace for DYAD
char *prod_managed_path; // producer path managed by DYAD
char *cons_managed_path; // consumer path managed by DYAD
bool
relative_to_managed_path; // relative path is relative to the managed path
// Internal
void *h; // the Flux handle for DYAD
struct dyad_dtl *dtl_handle; // Opaque handle to DTL info
const char *fname; // Used to track which file is getting processed.
bool use_fs_locks; // Used to track if fs locks should be used.
char *prod_real_path; // producer managed real path
char *cons_real_path; // consumer managed real path
uint32_t prod_managed_len; // length of producer path managed by DYAD
uint32_t cons_managed_len; // length of consumer path managed by DYAD
uint32_t prod_real_len; // length of producer managed real path
uint32_t cons_real_len; // length of consumer managed real path
uint32_t prod_managed_hash; // hash of producer path managed by DYAD
uint32_t cons_managed_hash; // hash of consumer path managed by DYAD
uint32_t prod_real_hash; // hash of producer managed real path
uint32_t cons_real_hash; // hash of consumer managed real path
uint32_t delim_len; // length of path delimiter
// User Facing
bool debug; // if true, perform debug logging
bool check; // if true, perform some check logging
bool reenter; // if false, do not recursively enter DYAD
bool initialized; // if true, DYAD is initialized
bool shared_storage; // if true, the managed path is shared
bool async_publish; // Enable asynchronous publish by producer
bool fsync_write; // Apply fsync after write by producer
unsigned int key_depth; // Depth of bins for the Flux KVS
unsigned int key_bins; // Number of bins for the Flux KVS
uint32_t rank; // Flux rank for DYAD
uint32_t service_mux; // Number of Flux brokers sharing node-local storage
uint32_t node_idx; // Index of the node hosting broker(s)
int pid; // unix process id, obtained by getpid()
char *kvs_namespace; // Flux KVS namespace for DYAD
char *prod_managed_path; // producer path managed by DYAD
char *cons_managed_path; // consumer path managed by DYAD
bool relative_to_managed_path; // relative path is relative to the managed path
};
typedef void *ucx_ep_cache_h;

#ifdef __cplusplus
}
#endif

#endif // DYAD_COMMON_STRUCTURES_INT_H
#endif // DYAD_COMMON_STRUCTURES_INT_H
573 changes: 288 additions & 285 deletions src/dyad/wrapper/wrapper.c
View file

Large diffs are not rendered by default.

0 comments on commit fc7ba50

Please sign in to comment.