Add new dirents_extended operation for pfind (c9eaf8e3) · Commits · David Auer / gekkofs

examples/gfind/gfind.cc

0 → 100644

+516 −0

Original line number	Diff line number	Diff line
		/*
		Copyright 2018-2020, Barcelona Supercomputing Center (BSC), Spain
		Copyright 2015-2020, Johannes Gutenberg Universitaet Mainz, Germany

		This software was partially supported by the
		EC H2020 funded project NEXTGenIO (Project ID: 671951, www.nextgenio.eu).

		This software was partially supported by the
		ADA-FS project under the SPPEXA project funded by the DFG.

		SPDX-License-Identifier: MIT
		*/

		/* Based on pfind from ior500 */
		/* https://github.com/VI4IO/pfind/ */

		#include <cmath>
		#include <cstring>
		#include <getopt.h>
		#include <iostream>
		#include <mpi.h>
		#include <queue>
		#include <regex.h>
		#include <stdio.h>
		#include <string>
		#include <sys/stat.h>
		#include <sys/types.h>
		#include <unistd.h>

		using namespace std;

		/* Minimal struct needed for io500 find */
		/* We could also do the filtering on the server */
		struct dirent_extended {
		size_t size;
		time_t ctime;
		unsigned short d_reclen;
		unsigned char d_type;
		char d_name[1];
		};

		/* Function exported from GekkoFS LD_PRELOAD, code needs to be compiled with
		* -fPIC */
		extern "C" int gkfs_getsingleserverdir(const char *path,
		struct dirent_extended *dirp,
		unsigned int count, int server)
		__attribute__((weak));

		/* PFIND OPTIONS EXTENDED We need to add the GekkoFS mount dir and the number of
		* servers */
		typedef struct {
		char *workdir;
		int just_count;
		int print_by_process;
		char *results_dir;
		int stonewall_timer;
		int print_rates;

		char *timestamp_file;
		char *name_pattern;
		regex_t name_regex;
		uint64_t size;

		int num_servers;
		char *mountdir;
		// optimizing parameters NOT USED
		int queue_length;
		int max_entries_per_iter;
		int steal_from_next; // if true, then steal from the next process
		int parallel_single_dir_access; // if 1, use hashing to parallelize single
		// directory access, if 2 sequential increment

		int verbosity;
		} pfind_options_t;

		typedef struct {
		uint64_t ctime_min;
		double stonewall_endtime;
		FILE *logfile;
		int needs_stat;
		} pfind_runtime_options_t;

		static pfind_runtime_options_t runtime;

		int pfind_rank;

		static pfind_options_t *opt;

		void pfind_abort(char *str) {
		printf("%s", str);
		exit(1);
		}

		static void pfind_print_help(pfind_options_t *res) {
		printf("pfind \nSynopsis:\n"
		"pfind <workdir> [-newer <timestamp file>] [-size <size>c] [-name "
		"<substr>] [-regex <regex>] [-S <numserver>] [-M <mountdir>]\n"
		"\tworkdir = \"%s\"\n"
		"\t-newer = \"%s\"\n"
		"\t-name\|-regex = \"%s\"\n"
		"\t-S: num servers = \"%s\"\n"
		"\t-M: mountdir = \"%s\"\n"
		"Optional flags\n"
		"\t-h: prints the help\n"
		"\t--help: prints the help without initializing MPI\n",
		res->workdir, res->timestamp_file, res->name_pattern, res->num_servers,
		res->mountdir);
		}
		MPI_Comm pfind_com;
		int pfind_size;
		pfind_options_t pfind_parse_args(int argc, char *argv, int force_print_help,
		MPI_Comm com) {
		MPI_Comm_rank(com, &pfind_rank);
		MPI_Comm_size(com, &pfind_size);
		pfind_com = com;

		pfind_options_t res = (pfind_options_t )malloc(sizeof(pfind_options_t));
		memset(res, 0, sizeof(pfind_options_t));
		int print_help = force_print_help;

		res->workdir = "./";
		res->results_dir = NULL;
		res->verbosity = 0;
		res->timestamp_file = NULL;
		res->name_pattern = NULL;
		res->size = UINT64_MAX;
		res->queue_length = 100000;
		res->max_entries_per_iter = 1000;
		char *firstarg = NULL;

		// when we find special args, we process them
		// but we need to replace them with 0 so that getopt will ignore them
		// and getopt will continue to process beyond them
		for (int i = 1; i < argc - 1; i++) {
		if (strcmp(argv[i], "-newer") == 0) {
		res->timestamp_file = strdup(argv[i + 1]);
		argv[i][0] = 0;
		argv[++i][0] = 0;
		} else if (strcmp(argv[i], "-size") == 0) {
		char *str = argv[i + 1];
		char extension = str[strlen(str) - 1];
		str[strlen(str) - 1] = 0;
		res->size = atoll(str);
		switch (extension) {
		case 'c':
		break;
		default:
		pfind_abort("Unsupported exension for -size\n");
		}
		argv[i][0] = 0;
		argv[++i][0] = 0;
		} else if (strcmp(argv[i], "-name") == 0) {
		res->name_pattern = (char )malloc(strlen(argv[i + 1]) 4 + 100);
		// transform a traditional name pattern to a regex:
		char *str = argv[i + 1];
		char *out = res->name_pattern;
		int pos = 0;
		for (unsigned i = 0; i < strlen(str); i++) {
		if (str[i] == '*') {
		pos += sprintf(out + pos, ".*");
		} else if (str[i] == '.') {
		pos += sprintf(out + pos, "[.]");
		} else if (str[i] == '"' \|\| str[i] == '\"') {
		// erase the "
		} else {
		out[pos] = str[i];
		pos++;
		}
		}
		out[pos] = 0;

		int ret = regcomp(&res->name_regex, res->name_pattern, 0);
		if (ret) {
		pfind_abort("Invalid regex for name given\n");
		}
		argv[i][0] = 0;
		argv[++i][0] = 0;
		} else if (strcmp(argv[i], "-regex") == 0) {
		res->name_pattern = strdup(argv[i + 1]);
		int ret = regcomp(&res->name_regex, res->name_pattern, 0);
		if (ret) {
		pfind_abort("Invalid regex for name given\n");
		}
		argv[i][0] = 0;
		argv[++i][0] = 0;
		} else if (!firstarg) {
		firstarg = strdup(argv[i]);
		argv[i][0] = 0;
		}
		}
		if (argc == 2) {
		firstarg = strdup(argv[1]);
		}

		int c;
		while ((c = getopt(argc, argv, "CPs:r:vhD:xq:H:NM:S:")) != -1) {
		if (c == -1) {
		break;
		}

		switch (c) {
		case 'H':
		res->parallel_single_dir_access = atoi(optarg);
		break;
		case 'N':
		res->steal_from_next = 1;
		break;
		case 'x':
		/* ignore fake arg that we added when we processed the extra args */
		break;
		case 'P':
		res->print_by_process = 1;
		break;
		case 'C':
		res->just_count = 1;
		break;
		case 'D':
		if (strcmp(optarg, "rates") == 0) {
		res->print_rates = 1;
		} else {
		pfind_abort("Unsupported debug flag\n");
		}
		break;
		case 'h':
		print_help = 1;
		break;
		case 'r':
		res->results_dir = strdup(optarg);
		break;
		case 'q':
		res->queue_length = atoi(optarg);
		break;
		if (res->queue_length < 10) {
		pfind_abort("Queue must be at least 10 elements!\n");
		}
		break;
		case 's':
		res->stonewall_timer = atol(optarg);
		break;
		case 'S':
		res->num_servers = atoi(optarg);
		break;
		case 'M':
		res->mountdir = strdup(optarg);
		break;
		case 'v':
		res->verbosity++;
		break;
		case 0:
		break;
		}
		}
		if (res->verbosity > 2 && pfind_rank == 0) {
		printf("Regex: %s\n", res->name_pattern);
		}

		if (print_help) {
		if (pfind_rank == 0)
		pfind_print_help(res);
		int init;
		MPI_Initialized(&init);
		if (init) {
		MPI_Finalize();
		}
		exit(0);
		}

		if (!firstarg) {
		pfind_abort("Error: pfind <directory>\n");
		}
		res->workdir = firstarg;

		return res;
		}

		/* Master send a new path to the workers */
		void send_newPath(string path) {
		int count = path.size() + 1;
		MPI_Bcast(&count, 1, MPI_INT, 0, MPI_COMM_WORLD);
		MPI_Bcast((void *)path.c_str(), count, MPI_CHAR, 0, MPI_COMM_WORLD);
		}

		/* Clients get a new path, getting a "0" size char means there is no new path*/
		string recv_newPath() {
		int count;
		MPI_Bcast(&count, 1, MPI_INT, 0, MPI_COMM_WORLD);
		if (count == 0)
		return "Terminate";
		char buf[count];
		MPI_Bcast(buf, count, MPI_CHAR, 0, MPI_COMM_WORLD);
		return buf;
		}

		/* Client Processing a path.
		* We increment local checked/found based on the filters
		* Each client sends the request to a subset of GekkoFS servers.
		* We use 102400 (plus space from 255 chars paths) so it is nearly 1M files per
		* server, which is enough for most cases
		*
		*/
		void dirProcess(const string path, unsigned long long &checked,
		unsigned long long &found, queue<string> &dirs,
		unsigned int world_rank, unsigned int world_size,
		pfind_options_t *opt) {
		struct dirent_extended getdir = (struct dirent_extended )malloc(
		(sizeof(struct dirent_extended) + 255) * 1024 * 100);
		memset(getdir, 0, (sizeof(struct dirent_extended) + 255) * 1024 * 100);
		// cout << "PROCESSING " << world_rank << "/"<< world_size << " = " << path <<
		// endl;

		for (int it = 0; it < opt->num_servers; it++) {
		int servers_per_node = ceil(opt->num_servers / (world_size - 1));
		if (servers_per_node == 0)
		servers_per_node++;
		for (int it = 0; it < servers_per_node; it++) {
		auto server = (world_rank - 1) * servers_per_node + it;
		if (server >= opt->num_servers)
		break;

		unsigned long long total_size = 0;
		auto n = gkfs_getsingleserverdir(
		path.c_str(), getdir,
		(sizeof(struct dirent_extended) + 255) * 1024 * 100, server);
		struct dirent_extended *temp = getdir;

		while (total_size < n) {
		if (strlen(temp->d_name) == 0)
		break;
		total_size += temp->d_reclen;
		/* Queue directory to process */
		if (temp->d_type == 1) {
		string slash;
		if (path[path.size() - 1] != '/')
		slash = "/";
		checked++;
		dirs.push(path + slash + temp->d_name);
		temp = reinterpret_cast<dirent_extended >((char )temp +
		temp->d_reclen);
		continue;
		}
		/* Find filtering */
		bool timeOK = true;
		if (opt->timestamp_file) {
		if ((uint64_t)temp->ctime < runtime.ctime_min)
		timeOK == false;
		}
		if (timeOK and (temp->size == opt->size or opt->size == UINT64_MAX))
		if (!(opt->name_pattern &&
		regexec(&opt->name_regex, temp->d_name, 0, NULL, 0)))
		found++;
		checked++;
		temp =
		reinterpret_cast<dirent_extended >((char )temp + temp->d_reclen);
		}
		}

		// cout << "/PROCESSING " << world_rank << "/" << world_size << " = " <<
		// path
		// << " Q " << dirs.size() << endl;
		}
		}
		int process(char *processor_name, int world_rank, int world_size,
		pfind_options_t *opt) {
		// Print off a hello world message

		// INIT PFIND
		memset(&runtime, 0, sizeof(pfind_runtime_options_t));
		int ret;
		/* Get timestamp file */
		if (opt->timestamp_file) {
		if (pfind_rank == 0) {
		static struct stat timer_file;
		if (lstat(opt->timestamp_file, &timer_file) != 0) {
		printf("Could not open: \"%s\", error: %s", opt->timestamp_file,
		strerror(errno));
		pfind_abort("\n");
		}
		runtime.ctime_min = timer_file.st_ctime;
		}
		MPI_Bcast(&runtime.ctime_min, 1, MPI_INT, 0, pfind_com);
		}

		int iterations = 0;
		if (world_rank == 0) {
		queue<string> dirs;
		string workdir = opt->workdir;
		workdir = workdir.substr(strlen(opt->mountdir), workdir.size());
		if (workdir.size() == 0)
		workdir = "/";
		dirs.push(workdir);

		do {

		string processpath = dirs.front();
		dirs.pop();
		// DISTRIBUTE WORK
		send_newPath(processpath);
		bool received_strings = true;
		// We need to gather new directories found (we use send-recv)
		for (int i = 1; i < world_size; i++) {
		received_strings = true;
		while (received_strings) {
		received_strings = false;
		// cout << " Checking from " << i << endl;
		MPI_Status mpistatus;
		MPI_Probe(i, 0, MPI_COMM_WORLD, &mpistatus);
		int count;
		MPI_Get_count(&mpistatus, MPI_CHAR, &count);
		char buf[count];
		MPI_Recv(&buf, count, MPI_CHAR, i, 0, MPI_COMM_WORLD, &mpistatus);
		if (count == 0) {
		continue;
		}
		// cout << " Receiving from " << i << " ---- " << buf << endl;
		string s = buf;
		dirs.push(s);
		received_strings = true;
		}
		}
		// cout << "NO more paths " << dirs.size() << endl;
		} while (!dirs.empty());

		int count = 0;
		MPI_Bcast(&count, 1, MPI_INT, 0, MPI_COMM_WORLD);

		MPI_Barrier(MPI_COMM_WORLD);

		unsigned long long *Array_checked =
		(unsigned long long )malloc(sizeof(unsigned long long) world_size);
		unsigned long long *Array_found =
		(unsigned long long )malloc(sizeof(unsigned long long) world_size);
		unsigned long long checked = 0;
		unsigned long long found = 0;

		MPI_Gather(&checked, 1, MPI_UNSIGNED_LONG_LONG, Array_checked, 1,
		MPI_UNSIGNED_LONG_LONG, 0, MPI_COMM_WORLD);
		MPI_Gather(&found, 1, MPI_UNSIGNED_LONG_LONG, Array_found, 1,
		MPI_UNSIGNED_LONG_LONG, 0, MPI_COMM_WORLD);

		for (int i = 0; i < world_size; i++) {
		checked += Array_checked[i];
		found += Array_found[i];
		}

		cout << "MATCHED " << found << "/" << checked << endl;
		}

		else {
		unsigned long long checked = 0;
		unsigned long long found = 0;
		while (1) {

		string toProcess = recv_newPath();
		if (toProcess == "Terminate") {
		break;
		}
		// cout << "REceived " << toProcess << " --- " << world_rank << endl;
		queue<string> dirs;

		dirProcess(toProcess, checked, found, dirs, world_rank, world_size, opt);
		// Send NEW DIRS to master
		while (!dirs.empty()) {
		string s = dirs.front();
		dirs.pop();
		// cout << world_rank << " --> Sending " << s << endl;
		MPI_Send((void *)s.c_str(), (s.size() + 1), MPI_CHAR, 0, 0,
		MPI_COMM_WORLD);
		}
		// cout << world_rank << " --> Sending 0 " << endl;
		MPI_Send((void *)0, 0, MPI_CHAR, 0, 0, MPI_COMM_WORLD);
		}

		MPI_Barrier(MPI_COMM_WORLD);
		MPI_Gather(&checked, 1, MPI_UNSIGNED_LONG_LONG, NULL, 1,
		MPI_UNSIGNED_LONG_LONG, 0, MPI_COMM_WORLD);
		MPI_Gather(&found, 1, MPI_UNSIGNED_LONG_LONG, NULL, 1,
		MPI_UNSIGNED_LONG_LONG, 0, MPI_COMM_WORLD);
		}

		return 0;
		}

		int main(int argc, char **argv) {

		for (int i = 0; i < argc; i++) {
		if (strcmp(argv[i], "--help") == 0) {
		argv[i][0] = 0;
		pfind_rank = 0;
		pfind_parse_args(argc, argv, 1, MPI_COMM_SELF);
		exit(0);
		}
		}

		// Initialize the MPI environment
		MPI_Init(&argc, &argv);

		// Get the number of processes
		int world_size;
		MPI_Comm_size(MPI_COMM_WORLD, &world_size);

		// Get the rank of the process
		int world_rank;
		MPI_Comm_rank(MPI_COMM_WORLD, &world_rank);

		opt = pfind_parse_args(argc, argv, 0, MPI_COMM_WORLD);
		// cout << opt->num_servers << " -- " << opt->mountdir << endl;
		// Get the name of the processor
		char processor_name[MPI_MAX_PROCESSOR_NAME];
		int name_len;
		MPI_Get_processor_name(processor_name, &name_len);

		process(processor_name, world_rank, world_size, opt);

		// Finalize the MPI environment.
		MPI_Finalize();
		}

examples/gfind/pfind.sh

0 → 100755

+6 −0

Original line number	Diff line number	Diff line
		#!/bin/bash
		# optimal $GKFS_FIND_PROCESS is $GKFS_SERVERS+1 as we align servers and find processes if the $NUM_NODES are the same
		# Output is saved to a file, so it can be processed by io500
		srun --nvram-options=1LM:1980 -N $NUM_NODES -n $GKFS_FIND_PROCESS --export=ALL,PSM2_DEVICES=self,hfi,shm,PSM2_MULTIRAIL=1,PSM2_MULTI_EP=0,LD_PRELOAD=${GKFS_PRLD} -o find_${SLURM_JOB_ID}.txt $GKFS_FIND $@ -M $GKFS_MNT -S $GKFS_SERVERS
		tail -n1 find_${SLURM_JOB_ID}.txt
		exit 0

examples/gfind/sfind.cc

0 → 100644

+383 −0

File added.

Preview size limit exceeded, changes collapsed.

include/client/gkfs_functions.hpp

+3 −1

Original line number	Diff line number	Diff line
		@@ -101,7 +101,9 @@ int gkfs_getdents64(unsigned int fd, struct linux_dirent64* dirp, unsigned int c

		int gkfs_rmdir(const std::string& path);

		//int gkfs_getsingleserverdir (const char * path, struct linux_dirent64* dirp, unsigned int count, int server);

		} // namespace syscall
		} // namespace gkfs

		extern "C" int gkfs_getsingleserverdir (const char * path, struct dirent_extended* dirp, unsigned int count, int server);
		#endif //GEKKOFS_GKFS_FUNCTIONS_HPP

include/client/rpc/forward_metadata.hpp

+3 −1

Original line number	Diff line number	Diff line
		@@ -17,7 +17,7 @@

		#include <string>
		#include <memory>

		#include <vector>
		/* Forward declaration */
		namespace gkfs {
		namespace filemap {
		@@ -51,6 +51,8 @@ std::pair<int, off64_t> forward_get_metadentry_size(const std::string& path);

		std::pair<int, std::shared_ptr<gkfs::filemap::OpenDir>> forward_get_dirents(const std::string& path);

		std::pair<int, std::vector<std::tuple<const std::string, bool, size_t, time_t>>> forward_get_dirents_single(const std::string& path, int server);

		#ifdef HAS_SYMLINKS

		int forward_mk_symlink(const std::string& path, const std::string& target_path);