diff --git a/fastlib/base/common.c b/fastlib/base/common.c index e7d9804e2a..7ebfe0711b 100644 --- a/fastlib/base/common.c +++ b/fastlib/base/common.c @@ -139,29 +139,35 @@ char *tsprintf(const char *format, ...) { } void percent_indicator(const char *what, uint64 numerator, uint64 denominator) { + static int last_percent = -1; char buf[80]; int length = 50; int i = 0; + int percent = (int)(numerator * 100 / denominator); - if (numerator == denominator) { - for (; i < sizeof(buf) - 1; i++) { - buf[i] = ' '; - } - buf[i] = '\0'; - fprintf(stderr, "%s\r", buf); - } else { - int percent = (int)(numerator * 100 / denominator); - - for (i = 0; i < percent * length / 100; i++) { - buf[i] = '#'; - } - for (; i < length; i++) { - buf[i] = '.'; + if (unlikely(percent != last_percent)) { + int num_ticks = percent * length / 100; + if (percent == 100) { + for (; i < sizeof(buf) - 1; i++) { + buf[i] = ' '; + } + buf[i] = '\0'; + fprintf(stderr, "%s\r", buf); + } else { + for (i = 0; i < num_ticks; i++) { + buf[i] = '#'; + } + for (; i < length; i++) { + buf[i] = '.'; + } + + buf[i] = '\0'; + + fprintf(stderr, "\r"ANSI_BLUE"STATUS: [%s] %02d%% %s"ANSI_CLEAR" \r", + buf, percent, what); } - buf[i] = '\0'; - - fprintf(stderr, ANSI_BLUE"STATUS: [%s] %02d%% %s"ANSI_CLEAR"\r", - buf, percent, what); + last_percent = percent; } + } diff --git a/fastlib/col/rangeset.h b/fastlib/col/rangeset.h index 8318df58c8..0075bbdf2f 100644 --- a/fastlib/col/rangeset.h +++ b/fastlib/col/rangeset.h @@ -101,7 +101,7 @@ class RangeSet { return ranges_[i]; } - const index_t size() const { + index_t size() const { return ranges_.size(); } diff --git a/fastlib/fx/datastore.c b/fastlib/fx/datastore.c index 487f9a5ce8..ff784b9884 100644 --- a/fastlib/fx/datastore.c +++ b/fastlib/fx/datastore.c @@ -192,6 +192,72 @@ struct datanode *datanode_get_paths(struct datanode *node, return node; } +static int datanode__unhex_char(char c) { + c = (c & ~0x20); + if (c <= '9') { + return c - '0'; + } else { + return c - 'A' + 10; + } +} + +static char *datanode__unhex(char *s) { + char *d = s; + while (*s) { + if (*s == '%' && isxdigit(s[1]) && isxdigit(s[2])) { + *d = datanode__unhex_char(s[1]) * 16 + datanode__unhex_char(s[2]); + s += 2; + } else { + *d = *s; + } + s++; + d++; + } + *d = '\0'; + return d; +} + +void datanode_read(struct datanode *node, nodetype_t type, FILE *stream) { + char buf[4096]; + + while (fgets(buf, sizeof(buf), stream) != NULL) { + int len; + char *key; + char *value; + struct datanode *result; + + // strip whitespace + for (len = strlen(buf); len >= 0 && isspace(buf[len-1]); len--) {} + buf[len] = '\0'; + key = buf; + + if (len == 0 || buf[0] == '#') { + // ignore blank lines and #comments + continue; + } + + value = strchr(key, ' '); + if (!value) { + NONFATAL("Unrecognized line: %s", buf); + break; + } + + *value = '\0'; + while (isspace(*++value)) {} + + datanode__unhex(key); + datanode__unhex(value); + + result = datanode_get_path(node, key, type); + if (result->val) { + NONFATAL("Overwriting [%s] - old value [%s], new [%s].", + key, (char*)result->val, value); + free(result->val); + } + result->val = strdup(value); + } +} + static char *datanode__hex(char *dest, const char *src, int remaining) { int c; @@ -233,7 +299,7 @@ static void datanode__write_buf_backwards(struct datanode *node, FILE *f, static void datanode__write_buf(struct datanode *node, FILE *f, char *prefix, char *buf) { - if (node->val) { + if (node->val) { buf[0] = ' '; datanode__hex(buf + 1, node->val, 4096 - (buf - prefix) - 1); fprintf(f, "%s\n", prefix); diff --git a/fastlib/fx/datastore.h b/fastlib/fx/datastore.h index ccce89d59a..adb9110926 100644 --- a/fastlib/fx/datastore.h +++ b/fastlib/fx/datastore.h @@ -96,6 +96,11 @@ struct datanode *datanode_get_paths(struct datanode *node, */ void datanode_write(struct datanode *node, FILE *f); +/** + * Populates a datanode from a file. + */ +void datanode_read(struct datanode *node, nodetype_t type, FILE *stream); + EXTERN_C_END #endif diff --git a/fastlib/fx/fx.c b/fastlib/fx/fx.c index 55bf2ccfb1..fd27f4d3b0 100644 --- a/fastlib/fx/fx.c +++ b/fastlib/fx/fx.c @@ -70,7 +70,17 @@ static void fx__parse_cmd_line(struct datanode *node, int argc, char *argv[]) *val++ = '\0'; } - datanode_get_path(node, path, NODETYPE_PARAM)->val = strdup(val); + if (strcmp(path, "fx/load") == 0) { + FILE *stream = fopen(val, "r"); + if (stream == NULL) { + FATAL("File not found (in --fx/load): [%s].", val); + } else { + datanode_read(node, NODETYPE_PARAM, stream); + (void)fclose(stream); + } + } else { + datanode_get_path(node, path, NODETYPE_PARAM)->val = strdup(val); + } free(s); } else { @@ -89,7 +99,7 @@ static void fx__read_debug_params(struct datanode *node) abort_on_nonfatal = fx_param_bool(node, "./abort_on_nonfatal", 0); pause_on_nonfatal = fx_param_bool(node, "./pause_on_nonfatal", 0); print_notify_headers = fx_param_bool(node, "./print_notify_headers", 1); - fx__show_results_timers = !(fx_param_bool(node, "./quiet", 0)); + fx__show_results_timers = fx_param_bool(node, "./noisy", 0); } static void fx__attempt_speedup() diff --git a/fastlib/u/garryb/nbr/affinity.cc b/fastlib/u/garryb/nbr/affinity.cc index c6c164c7e3..f533720fe0 100644 --- a/fastlib/u/garryb/nbr/affinity.cc +++ b/fastlib/u/garryb/nbr/affinity.cc @@ -698,7 +698,14 @@ void AffinityMain(datanode *module, const char *gnp_name) { const int RHO_CHANNEL = 360; const int REDUCE_CHANNEL = 370; const int DONE_CHANNEL = 390; + int convergence = fx_param_int(module, "affinity/convergence", 30); int stable_iterations = 0; + int maxit = fx_param_int(module, "affinity/maxit", 1000); + + if (!rpc::is_root()) { + // turn off fastexec output + fx_silence(); + } timestats.Init(); @@ -712,8 +719,8 @@ void AffinityMain(datanode *module, const char *gnp_name) { // One thing to note: alpha and rho are never taking up // RAM at the same time! - size_t alpha_mb = fx_param_int(module, "alpha_mb", 200); - size_t rho_mb = fx_param_int(module, "rho_mb", 100); + size_t alpha_mb = fx_param_int(module, "alpha/mb", 200); + size_t rho_mb = fx_param_int(module, "rho/mb", 100); timer *timer_alpha = fx_timer(module, "all_alpha"); timer *timer_rho = fx_timer(module, "all_rho"); @@ -737,7 +744,9 @@ void AffinityMain(datanode *module, const char *gnp_name) { index_t n_points = tree.n_points(); - for (int iter = 0;; iter++) { + for (int iter = 0;;) { + iter++; + fx_timer_start(module, "all_alpha"); thor_utils::RpcDualTree >( fx_submodule(module, "thor", "iter/%d/alpha", iter), 200, @@ -748,8 +757,7 @@ void AffinityMain(datanode *module, const char *gnp_name) { rpc::Reduce(REDUCE_CHANNEL+0, VisitorReductor(), &apply_alphas); if (rpc::is_root()) { fprintf(stderr, ANSI_RED"--- %3d: alpha: max1=%f, max2=%f"ANSI_CLEAR"\n", - iter, - apply_alphas.sum_alpha1 / n_points, + iter, apply_alphas.sum_alpha1 / n_points, apply_alphas.sum_alpha2 / n_points); } alphas.ResetElements(); @@ -785,7 +793,7 @@ void AffinityMain(datanode *module, const char *gnp_name) { } else { stable_iterations = 0; } - done.SetData(stable_iterations >= 16); + done.SetData(iter >= maxit || stable_iterations >= convergence); } done.Doit(DONE_CHANNEL); @@ -801,9 +809,23 @@ void AffinityMain(datanode *module, const char *gnp_name) { } +int main(int argc, char *argv[]) { + fx_init(argc, argv); + rpc::Init(); + + srand(time(NULL)); + + AffinityMain(fx_root, "affinity"); + + rpc::Done(); + fx_done(); +} + #if 0 +// The old source code! + // // AffinityAlpha::Param param; // @@ -1213,15 +1235,3 @@ void AffinityMain(datanode *module, const char *gnp_name) { //} //*/ #endif - -int main(int argc, char *argv[]) { - fx_init(argc, argv); - rpc::Init(); - - srand(time(NULL)); - - AffinityMain(fx_root, "affinity"); - - rpc::Done(); - fx_done(); -} diff --git a/fastlib/u/garryb/nbr/blockdev.cc b/fastlib/u/garryb/nbr/blockdev.cc index 5aa297c6c6..047f5100dc 100644 --- a/fastlib/u/garryb/nbr/blockdev.cc +++ b/fastlib/u/garryb/nbr/blockdev.cc @@ -220,9 +220,9 @@ MemBlockDevice::~MemBlockDevice() { void IoStats::Report(datanode *module) const { fx_format_result(module, "n_reads", "%u", n_reads_); - fx_format_result(module, "n_read_bytes", "%"L64"u", n_read_bytes_); + //fx_format_result(module, "n_read_bytes", "%"L64"u", n_read_bytes_); fx_format_result(module, "n_writes", "%u", n_writes_); - fx_format_result(module, "n_write_bytes", "%"L64"u", n_write_bytes_); + //fx_format_result(module, "n_write_bytes", "%"L64"u", n_write_bytes_); } void IoStats::Report(BlockDevice::offset_t n_block_bytes, @@ -235,4 +235,5 @@ void IoStats::Report(BlockDevice::offset_t n_block_bytes, fx_format_result(module, "write_ratio", "%.4f", n_write_bytes_ / (double(n_block_bytes) * n_blocks)); fx_format_result(module, "n_block_bytes", "%u", n_block_bytes); + fx_format_result(module, "n_blocks", "%u", n_blocks); } diff --git a/fastlib/u/garryb/nbr/distribcache.cc b/fastlib/u/garryb/nbr/distribcache.cc index ae43d13ad5..4f6558933d 100644 --- a/fastlib/u/garryb/nbr/distribcache.cc +++ b/fastlib/u/garryb/nbr/distribcache.cc @@ -42,10 +42,6 @@ void DistributedCache::InitMaster(int channel_num_in, void DistributedCache::InitWorker( int channel_num_in, size_t total_ram, BlockHandler *handler_in) { - if (total_ram < 65536) { - FATAL("total_ram for a cache is unusually low (%ld) -- remember this is in bytes!", - long(total_ram)); - } InitCommon_(); // connect to master and figure out specs ConfigTransaction ct; @@ -103,6 +99,10 @@ void DistributedCache::InitCommon_() { } void DistributedCache::InitCache_(size_t total_ram) { + if (total_ram < 65536) { + FATAL("total_ram for a cache is unusually low (%ld BYTES)!", + long(total_ram)); + } // give enough cache sets, but rounded up n_sets_ = (total_ram + ASSOC*n_block_bytes_ - 1) / (ASSOC*n_block_bytes_); slots_.Init(n_sets_ << LOG_ASSOC); @@ -252,14 +252,17 @@ void DistributedCache::StartSync() { void DistributedCache::WaitSync(datanode *node) { channel_.WaitSync(); if (node) { - disk_stats().Report(n_block_bytes_, n_blocks_, + /*disk_stats().Report(n_block_bytes_, n_blocks_, fx_submodule(node, NULL, "disk_stats")); net_stats().Report(n_block_bytes_, n_blocks_, - fx_submodule(node, NULL, "net_stats")); + fx_submodule(node, NULL, "net_stats"));*/ world_disk_stats().Report(n_block_bytes_, n_blocks_, fx_submodule(node, NULL, "world_disk_stats")); - world_net_stats().Report(n_block_bytes_, n_blocks_, - fx_submodule(node, NULL, "world_net_stats")); + if (rpc::n_peers() > 1) { + // net stats are only interesting if there's at least two machines + world_net_stats().Report(n_block_bytes_, n_blocks_, + fx_submodule(node, NULL, "world_net_stats")); + } } disk_stats_.Reset(); net_stats_.Reset(); @@ -403,12 +406,16 @@ void DistributedCache::HandleRemoteOwner_(blockid_t block, blockid_t end) { mutex_.Unlock(); } -void DistributedCache::GiveOwnership(blockid_t my_blockid, int new_owner) { +void DistributedCache::GiveOwnership(blockid_t blockid, int new_owner) { if (likely(new_owner != my_rank_)) { // mark whole block as dirty and change its owner. - StartWrite(my_blockid, false); mutex_.Lock(); - BlockMetadata *block = &blocks_[my_blockid]; + BlockMetadata *block = &blocks_[blockid]; + if (unlikely(block->locks == 0)) { + DecacheBlock_(blockid); + block->locks = 0; + } + block->status = FULLY_DIRTY; DEBUG_ASSERT_MSG(block->is_owner(), "Can only give ownership if I'm the owner"); if (block->local_blockid() != SELF_OWNER_UNALLOCATED) { @@ -418,9 +425,10 @@ void DistributedCache::GiveOwnership(blockid_t my_blockid, int new_owner) { overflow_free_ = block->local_blockid(); } block->value = ~new_owner; - DEBUG_ASSERT(block->status == FULLY_DIRTY); // set by StartWrite + if (unlikely(block->locks == 0)) { + EncacheBlock_(blockid); + } mutex_.Unlock(); - StopWrite(my_blockid); } } @@ -528,8 +536,8 @@ void DistributedCache::HandleLocalMiss_(BlockDevice::blockid_t blockid) { DEBUG_ASSERT(block->is_owner()); block->data = mem::Alloc(n_block_bytes_); - fprintf(stderr, "DISK: reading %d from %d (%d bytes)\n", - blockid, local_blockid, n_block_bytes_); + //fprintf(stderr, "DISK: reading %d from %d (%d bytes)\n", + // blockid, local_blockid, n_block_bytes_); disk_stats_.RecordRead(n_block_bytes_); overflow_device_->Read(local_blockid, 0, n_block_bytes_, block->data); @@ -594,7 +602,8 @@ void DistributedCache::EncacheBlock_(BlockDevice::blockid_t blockid) { if (!blocks_[base_slot[i].blockid].is_owner()) { break; } - if (--i = remote_preference) { + i--; + if (i == remote_preference) { i = ASSOC-1; break; } @@ -655,8 +664,8 @@ void DistributedCache::WritebackDirtyLocalFreeze_( DEBUG_ASSERT(block->is_dirty()); handler_->BlockFreeze(blockid, 0, n_block_bytes_, block->data, block->data); - fprintf(stderr, "DISK: writing %d to %d (%d bytes)\n", - blockid, local_blockid, n_block_bytes_); + //fprintf(stderr, "DISK: writing %d to %d (%d bytes)\n", + // blockid, local_blockid, n_block_bytes_); disk_stats_.RecordWrite(n_block_bytes_); overflow_device_->Write(local_blockid, 0, n_block_bytes_, block->data); block->status = NOT_DIRTY_OLD; diff --git a/fastlib/u/garryb/nbr/rpc_sock.cc b/fastlib/u/garryb/nbr/rpc_sock.cc index be0c479bd1..fa25ff30c8 100644 --- a/fastlib/u/garryb/nbr/rpc_sock.cc +++ b/fastlib/u/garryb/nbr/rpc_sock.cc @@ -799,7 +799,7 @@ void SockConnection::OpenOutgoing(bool blocking) { sizeof(struct sockaddr_in))) { (void) close(temp_fd); - if (elapsed_time % 10 == 0) { + if (elapsed_time % 10 == 0 && elapsed_time >= 10) { NONFATAL( "rpc_sock(%d): Connection to parent %d failed, we'll try for %d more seconds.\n", rpc::rank(), peer_, TIMEOUT_CONNECT - elapsed_time); diff --git a/fastlib/u/garryb/nbr/rpc_sock.h b/fastlib/u/garryb/nbr/rpc_sock.h index 4b184726c8..24cfaa4edf 100644 --- a/fastlib/u/garryb/nbr/rpc_sock.h +++ b/fastlib/u/garryb/nbr/rpc_sock.h @@ -459,11 +459,11 @@ namespace rpc { return RpcSockImpl::instance->n_peers(); } /** Get the i'th child. */ - inline const int child(int i) { + inline int child(int i) { return RpcSockImpl::instance->children()[i]; } /** Number of broadcast-tree children. */ - inline const index_t n_children() { + inline index_t n_children() { return RpcSockImpl::instance->children().size(); } /** Whether the root of the tree. */