Changeset 384

Timestamp:

08/29/06 19:30:39 (2 years ago)

Author:

bradfitz

Message:

merging the server-newrepl branch into trunk, now that it's pretty
stable and in use on livejournal.com.

Files:

• trunk/server/TODO (modified) (1 diff)
• trunk/server/doc/fsck-notes.txt (copied) (copied from branches/server-newrepl/doc/fsck-notes.txt)
• trunk/server/lib/MogileFS/Class.pm (copied) (copied from branches/server-newrepl/lib/MogileFS/Class.pm)
• trunk/server/lib/MogileFS/Config.pm (modified) (4 diffs)
• trunk/server/lib/MogileFS/ProcManager.pm (modified) (5 diffs)
• trunk/server/lib/MogileFS/ReplicationPolicy/MultipleHosts.pm (modified) (3 diffs)
• trunk/server/lib/MogileFS/Util.pm (modified) (1 diff)
• trunk/server/lib/MogileFS/Worker.pm (modified) (7 diffs)
• trunk/server/lib/MogileFS/Worker/Checker.pm (copied) (copied from branches/server-newrepl/lib/MogileFS/Worker/Checker.pm)
• trunk/server/lib/MogileFS/Worker/Monitor.pm (modified) (8 diffs)
• trunk/server/lib/MogileFS/Worker/Query.pm (modified) (19 diffs)
• trunk/server/lib/MogileFS/Worker/Replicate.pm (modified) (16 diffs)
• trunk/server/mogdbsetup (modified) (2 diffs)
• trunk/server/mogilefsd (modified) (13 diffs)
• trunk/server/mogstored (modified) (3 diffs)
• trunk/server/t/00-startup.t (modified) (3 diffs)
• trunk/server/t/lib/mogtestlib.pl (modified) (6 diffs)

trunk/server/TODO

@@ +1 @@
+-- if create_open before monitor runs yet, block and wait for a round?  better than 'no_devices'
+   might need some marker for 'end of monitoring'?  or just wait until we have 1 or 3?
+   or if they asked for 3, only return 1 in that rare case, preferring latency to redundancy.
+   plus, we'd know that 1 is writable in last few seconds!
+
+-- update the 'repl' command for new file_to_replicate table
+
--- replication policy error storm when a device is known to be observably down:
-

trunk/server/lib/MogileFS/Config.pm

@@ -34 +34 @@
-    $reaper_jobs,
-    $monitor_jobs,
+    $checker_jobs,
-    $mog_root,
-    $min_free_space,
@@ -65 +66 @@
-                             'default_mindevcount=i' => \$cmdline{default_mindevcount},
-                             'node_timeout=i' => \$cmdline{node_timeout},
+                             'no_schema_check' => \$cmdline{no_schema_check},
-                             );
-
@@ -117 +119 @@
-    $reaper_jobs    = choose_value( 'reaper_jobs', 1 );
-    $monitor_jobs   = choose_value( 'monitor_jobs', 1 );
+    $checker_jobs   = choose_value( 'checker_jobs', 1 );
-    $min_free_space = choose_value( 'min_free_space', 100 );
-    $max_disk_age   = choose_value( 'max_disk_age', 5 );
-
+$ENV{DEBUG} || 
-    $USE_HTTP       = ! choose_value( 'no_http', 0, 1);
-    $default_mindevcount = choose_value( 'default_mindevcount', 2 );
-    $node_timeout   = choose_value( 'node_timeout', 2 );
+
+    choose_value( 'no_schema_check', 0 );
-
-    # now load plugins
@@ -130 +135 @@
-}
-
-[, $boolean]
-($$;$) 
-, $boolean
+
+
+
-    return set_config($name, $cmdline{$name}) if defined $cmdline{$name};
-    return set_config($name, $cfgfile{$name}) if defined $cfgfile{$name};

trunk/server/lib/MogileFS/ProcManager.pm

@@ -50 +50 @@
-    my ($class, $job) = @_;
-    return {
+        checker     => "Checker",
-        queryworker => "Query",
-        delete      => "Delete",
@@ -398 +399 @@
-    return MogileFS::ProcManager->HandleChildRequest($worker, $line) if !$client;
-
+    # FIXME: HOW IS THIS EVER CALLED?  things with colons never go to HandleQueryWorkerResponse.
+    #        see MogileFS::Connection::Worker
-    # out-of-band messages (not replies to requests) start with a colon:
-    if ($line =~ /^:state_change (\w+) (\d+) (\w+)/) {
@@ -461 +464 @@
-        push @$clref, Time::HiRes::gettimeofday();
-        $Mappings{$worker->{fd}} = $clref;
+        $Stats{queries}++;
-
-        # increment our counter so we know what request counter this is going out
@@ -547 +551 @@
-
-    } elsif ($cmd eq ":still_alive") {
-
-        # a no-op
+
+    } elsif ($cmd eq ":monitor_just_ran") {
+        send_monitor_has_run($child);
+
+    } elsif ($cmd =~ /^:invalidate_meta (\w+)/) {
+        send_invalidate($1, $child);
-
-    } else {
@@ -628 +637 @@
-    #warn "STATE CHANGE: $what<$whatid> = $state\n";
-    # TODO: can probably send this to all children now, not just certain types
-
+ monitor checker
-        MogileFS::ProcManager->ImmediateSendToChildrenByJob($type, ":state_change $what $whatid $state", $child);
+    }
+}
+
+sub send_invalidate {
+    my ($what, $child) = @_;
+    # TODO: can probably send this to all children now, not just certain types
+    for my $type (qw(queryworker replicate delete monitor checker)) {
+        MogileFS::ProcManager->ImmediateSendToChildrenByJob($type, ":invalidate_meta_once $what", $child);
+    }
+}
+
+sub send_monitor_has_run {
+    my $child = shift;
+    for my $type (qw(replicate checker)) {
+        MogileFS::ProcManager->ImmediateSendToChildrenByJob($type, ":monitor_has_run", $child);
-    }
-}

trunk/server/lib/MogileFS/ReplicationPolicy/MultipleHosts.pm

@@ -8 +8 @@
-sub replicate_to {
-    my ($class, %args) = @_;
+
-    my $fid      = delete $args{fid};      # fid scalar to copy
-    my $on_devs  = delete $args{on_devs};  # arrayref of device objects
@@ -20 +21 @@
-    my $already_on = @$on_devs;
-
+    # FIXME: this is NOT true.  make sure it's on 2+ hosts at least, if min > 1.
+
-    # replication good.
-    return 0 if $already_on >= $min;
-
-    # see which and how many unique hosts we're already on.
+    my %on_dev;
-    my %on_host;
-    foreach my $dev (@$on_devs) {
-        $on_host{$dev->{hostid}} = 1;
+        $on_dev{$dev->{devid}} = 1;
-    }
-    my $uniq_hosts_on    = scalar keys %on_host;
@@ -38 +43 @@
-    }
-
-
+&& ! $on_dev{$_} 
-            Mgd::find_deviceid(
-                               random         => 1,

trunk/server/lib/MogileFS/Util.pm

@@ -12 +12 @@
-    while (1) {
-        my $start = Time::HiRes::time();
+        my $explicit_sleep = undef;
-
-        # run the code in a loop, so "next" will get out of it.
-        foreach (1) {
-
+
+
+
-        }
-
-        my $took = Time::HiRes::time() - $start;
-        my $sleep_for = $delay - $took;
-
+
+
+
+
+
-    }
-}

trunk/server/lib/MogileFS/Worker.pm

@@ -4 +4 @@
-            'last_bcast_state',   # "{device|host}-$devid" => [$time, {alive|dead}]
-            'readbuf',            # unparsed data from parent
+            'monitor_has_run',    # true once we've heard of the monitor job being alive
-            );
-
@@ -19 +20 @@
-    $self->{readbuf}          = '';
-    $self->{last_bcast_state} = {};
+    $self->{monitor_has_run}  = 0;
-
-    IO::Handle::blocking($psock, 0);
@@ -30 +32 @@
-sub get_dbh {
-    return Mgd::get_dbh();
+}
+
+sub monitor_has_run {
+    my $self = shift;
+    return $self->{monitor_has_run} ? 1 : 0;
-}
-
@@ -46 +53 @@
-    return 1 if $rv == $totallen;
-    die "Error writing: $!" if $!;
-
+
-    my $remain = $totallen - $rv;
-    my $offset = $rv;
@@ -82 +89 @@
-        my $rv = sysread($psock, $buf, 1024);
-        $self->{readbuf} .= $buf;
-
+
-        while ($self->{readbuf} =~ s/^(.+?)\r?\n//) {
-            my $line = $1;
@@ -148 +155 @@
-}
-
+sub invalidate_meta {
+    my ($self, $what) = @_;
+    return if $Mgd::INVALIDATE_NO_PROPOGATE;  # anti recursion
+    $self->send_to_parent(":invalidate_meta $what");
+}
+
-# tries to parse generic (not job-specific) commands sent from parent
-# to child.  returns 1 on success, or 0 if comman given isn't generic,
@@ -172 +185 @@
-    }
-
+    if ($$lineref =~ /^:invalidate_meta_once (\w+)/) {
+        local $Mgd::INVALIDATE_NO_PROPOGATE = 1;
+        eval("Mgd::invalidate_${1}_cache()");
+        return 1;
+    }
+
+    if ($$lineref =~ /^:monitor_has_run/) {
+        $self->{monitor_has_run} = 1;
+        return 1;
+    }
+
-    # TODO: warn on unknown commands?
-

trunk/server/lib/MogileFS/Worker/Monitor.pm

@@ -1 +1 @@
-package MogileFS::Worker::Monitor;
-# deletes files
-
-use strict;
@@ -46 +45 @@
-
-            my $host = $Mgd::cache_host{$dev->{hostid}};
-
-
+
+
+
-
-            # now try to get the data with a short timeout
@@ -61 +61 @@
-                if ($failed_after < 0.5) {
-                    $self->broadcast_device_unreachable($dev->{devid});
-
+get_
-                } else {
-                    $failed_after = sprintf("%.02f", $failed_after);
@@ -70 +70 @@
-                next;
-            }
+
+            # at this point we can reach the host
+            $self->broadcast_host_reachable($dev->{hostid});
-
-            my %stats;
@@ -107 +110 @@
-EOREQUEST
-
-            # TODO: hosts aren't writable.  they're "available"
-            # TODO: re-check the file was written as put.
-            # TODO: put something unique in the file
@@ -115 +117 @@
-            my $resp = $ua->request($req);
-            if ($resp->is_success) {
-                $self->broadcast_host_reachable($dev->{hostid});
-                $self->broadcast_device_writeable($dev->{devid});
-                error("dev$dev->{devid}: used = $used, total = $total, writeable = 1")
@@ -121 +122 @@
-            } else {
-                # merely readable
-                $self->broadcast_host_reachable($dev->{hostid});
-                $self->broadcast_device_readable($dev->{devid});
-                error("dev$dev->{devid}: used = $used, total = $total, writeable = 0")
@@ -127 +127 @@
-            }
-        }
+
+        # announce to the parent that we've run
+        $self->send_to_parent(":monitor_just_ran");
-    });
-

trunk/server/lib/MogileFS/Worker/Query.pm

@@ -51 +51 @@
-            next;
-        }
-
+
-        my $newread;
-        my $rv = sysread($psock, $newread, 1024);
@@ -127 +127 @@
-    my $args = shift;
-
-
-
-
-
-
-
+
+
+
+
-
-    return $self->ok_line;
@@ -160 +158 @@
-
-    # figure out what classid this file is for
-
+ || ""
-    my $classid = 0;
-    if (length($class)) {
@@ -173 +171 @@
-    my (@dests, @hosts);
-    my $devs = Mgd::get_device_summary();
+
-    while (scalar(@dests) < ($multi ? 3 : 1)) {
-        my $devid = Mgd::find_deviceid(
-random
-weight_by_fre
-not_on_hosts => \@hosts
-);
-
+                           random          
+                           must_be_writeabl
+                           weight_by_free   => 1
+                               not_on_hosts     => \@hosts,
+
-        last unless defined $devid;
-
@@ -292 +291 @@
-    # if a temp file is closed without a provided-key, that means to
-    # delete it.
-
+defined $key && 
-        # add to to-delete list
-        $dbh->do("REPLACE INTO file_to_delete SET fid=?", undef, $fid);
@@ -326 +325 @@
-                      $fid, $dmid, $key, $size, $trow->{classid});
-    return $self->err_line("db_error") unless $rv;
+
+    # mark it as needing replicating:
+    $dbh->do("INSERT IGNORE INTO file_to_replicate ".
+             "SET fid=?, fromdevid=?, nexttry=0", undef, $fid, $devid);
+    return $self->err_line("db_error") if $dbh->err;
-
-    $dbh->do("DELETE FROM tempfile WHERE fid=?", undef, $fid);
@@ -549 +553 @@
-        return $self->err_line("unknown_hostid") unless $Mgd::cache_host{$hostid};
-    } elsif (my $host = $args->{hostname}) {
-
-
-
-
-
-
+
-        return $self->err_line("unknown_host") unless $hostid;
-    }
@@ -563 +562 @@
-        return $self->err_line("existing_devid");
-    }
+    Mgd::invalidate_device_cache();
-    return $self->ok_line;
-}
@@ -588 +588 @@
-
-    # return the domain id we created
+    Mgd::invalidate_domain_cache();
-    return $self->ok_line({ domain => $domain });
-}
@@ -620 +621 @@
-
-    # return the domain we nuked
+    Mgd::invalidate_domain_cache();
-    return $self->ok_line({ domain => $domain });
-}
@@ -668 +670 @@
-
-    # return success
+    Mgd::invalidate_class_cache();
-    return $self->ok_line({ class => $class, mindevcount => $mindevcount, domain => $domain });
-}
@@ -709 +712 @@
-
-    # return the class we nuked
+    Mgd::invalidate_class_cache();
-    return $self->ok_line({ domain => $domain, class => $class });
-}
@@ -755 +759 @@
-    } else {
-        # get the max host id in use (FIXME: racy!)
-$dbh->selectrow_array('SELECT MAX(hostid) FROM host'
+($dbh->selectrow_array('SELECT MAX(hostid) FROM host') || 0
-
-        # now insert the new host
@@ -943 +947 @@
-
-    # success, weight changed
+    Mgd::invalidate_device_cache();
-    return $self->ok_line($ret);
-}
@@ -979 +984 @@
-
-    # success, state changed
+    Mgd::invalidate_device_cache();
-    return $self->ok_line($ret);
-}
@@ -1026 +1032 @@
-        }
-        $ret->{"replicationcount"} = $count;
+
+        # now we want to do the "new" replication stats
+        my $db_time = $dbh->selectrow_array('SELECT UNIX_TIMESTAMP()');
+        $stats = $dbh->selectall_arrayref('SELECT nexttry, COUNT(*) FROM file_to_replicate GROUP BY 1');
+        foreach my $stat (@$stats) {
+            if ($stat->[0] < 1000) {
+                # anything under 1000 is a specific state, so let's define those.  here's the list
+                # of short names to describe them.
+                my $name = {
+                    0 => 'newfile', # new files that need to be replicated
+                    1 => 'redo',    # files that need to go through replication again
+                }->{$stat->[0]} || "unknown";
+
+                # now put it in the output hashref.  note that we do += because we might
+                # have more than one group of unknowns.
+                $ret->{"to_replicate_$name"} += $stat->[1];
+
+            } elsif ($stat->[0] == MogileFS::Worker::Replicate::end_of_time()) {
+                $ret->{"to_replicate_manually"} = $stat->[1];
+
+            } elsif ($stat->[0] < $db_time) {
+                $ret->{"to_replicate_overdue"} += $stat->[1];
+
+            } else {
+                $ret->{"to_replicate_deferred"} += $stat->[1];
+            }
+        }
-    }
-
@@ -1072 +1105 @@
-}
-
+sub cmd_replicate_now {
+    my MogileFS::Worker::Query $self = shift;
+
+    my $dbh = Mgd::get_dbh()
+        or return $self->err_line('nodb');
+    my $rv = $dbh->do("UPDATE file_to_replicate SET nexttry = UNIX_TIMESTAMP() WHERE nexttry > UNIX_TIMESTAMP()");
+
+    return $self->err_line('db', $dbh->errstr) if $dbh->err;
+    return $self->ok_line({ count => int($rv) });
+}
+
+sub cmd_checker {
+    my MogileFS::Worker::Query $self = shift;
+    my $args = shift;
+
+    my $new_setting;
+    if ($args->{disable}) {
+        $new_setting = 'off';
+    } elsif ($args->{level}) {
+        # they want to turn it on or change the level, so let's ensure they
+        # specified a valid level
+        if (MogileFS::Worker::Checker::is_valid_level($args->{level})) {
+            $new_setting = $args->{level};
+        } else {
+            return $self->err_line('invalid_checker_level');
+        }
+    }
+
+    if (defined $new_setting) {
+        Mgd::set_server_setting('fsck_enable', $new_setting);
+        return $self->ok_line;
+    }
+
+    $self->err_line('failure');
+}
+
-sub ok_line {
-    my MogileFS::Worker::Query $self = shift;
@@ -1100 +1169 @@
-        'class_has_files' => "Class still has files, uanble to delete",
-        'class_not_found' => "Class not found",
+        'db' => "Database error",
-        'domain_has_files' => "Domain still has files, uanble to delete",
-        'domain_exists' => "That domain already exists",
@@ -1110 +1180 @@
-        'host_not_found' => "Host not found",
-        'invalid_chars' => "Patterns must not contain backslashes (\\) or percent signs (%).",
+        'invalid_checker_level' => "Checker level invalid.  Please see documentation on this command.",
-        'invalid_mindevcount' => "The mindevcount must be at least 1",
-        'key_exists' => "Target key name already exists; can't overwrite.",

trunk/server/lib/MogileFS/Worker/Replicate.pm

@@ -10 +10 @@
-use List::Util ();
-use MogileFS::Util qw(error every);
+use MogileFS::Class;
-use POSIX ":sys_wait_h"; # argument for waitpid
-use POSIX;
+
+# setup the value used in a 'nexttry' field to indicate that this item will never
+# actually be tried again and require some sort of manual intervention.
+use constant ENDOFTIME => 2147483647;
+
+sub end_of_time { ENDOFTIME; }
-
-sub new {
@@ -32 +39 @@
-}
-
+# { fid => lastcheck }; instructs us not to replicate this fid... we will clear
+# out fids from this list that are expired
+my %fidfailure;
+
+# { fid => 1 }; used to keep track of fids we find in the unreachable_fids table
+my %unreachable;
+my $dbh;
+
-sub work {
-    my $self = shift;
-
-
-
-
-
-
-
+
+
-
-    every(2.0, sub {
-        $self->parent_ping;
-
+        # replication doesn't go well if the monitor job hasn't actively started
+        # marking things as being available
+        unless ($self->monitor_has_run) {
+            error("waiting for monitor job to complete a cycle before beginning replication")
+                if time() > $warn_after;
+            return;
+        }
+
-        $self->validate_dbh;
-my 
+
-
-        # update our unreachable fid list... we consider them good for 15 minutes
@@ -62 +81 @@
-        }
-
-
-
-
-
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
-                if $Mgd::DEBUG >= 1;
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
-                    } else {
-
-
+
-                    }
-                }
+
+                $self->read_from_parent;
+
+                if (my $status = replicate($dbh, $fid, class => $mclass)) {
+                    # $status is either 0 (failure, handled below), 1 (success, we actually
+                    # replicated this file), or 2 (success, but someone else replicated it).
+                    # so if it's 2, we just want to go to the next fid.  this file is done.
+                    next if $status == 2;
+
+                    # if it was no longer reachable, mark it reachable
+                    if (delete $unreachable{$fid}) {
+                        $dbh->do("DELETE FROM unreachable_fids WHERE fid = ?", undef, $fid);
+                        die $dbh->errstr if $dbh->err;
+                    }
+
+                    # housekeeping
+                    $fixed++;
+                    $self->send_to_parent("repl_i_did $fid");
+
+                    # status update
+                    if ($fixed % 20 == 0) {
+                        my $ratio = $fixed/$attempted*100;
+                        error(sprintf("replicated=$fixed, attempted=$attempted, ratio=%.2f%%", $ratio))
+                            if $fixed % 20 == 0;
+                    }
+                } else {
+                    # failed in replicate, don't retry for a minute
+                    $fidfailure{$fid} = $now + 60;
+                }
-            }
-);
+
-    });
-}
-
-
+
+
+
+
+
+
-sub replicate {
-
-
+
+
+
+
+
+
+
+
+
+
+
+
+
-    eval "use $policy_class; 1;";
-    if ($@) {
@@ -154 +315 @@
-    }
-
+    my $lock;  # bool: whether we got the lock or not
-    my $lockname = "mgfs:fid:$fid:replicate";
-
-
-
-
+
+
+
+
+
+
+