hi

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;
   }
+
 }

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();
   }
 

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);

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

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()

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<AffinityAlpha, DualTreeDepthFirst<AffinityAlpha> >(
         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<ApplyAlphas>(), &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();
-}

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);
 }

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<char>(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;

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);

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. */