File Coverage

blib/lib/MogileFS/Connection/Client.pm
Criterion Covered Total %
statement 12 81 14.8
branch 0 40 0.0
condition 0 3 0.0
subroutine 4 13 30.7
pod 5 7 71.4
total 21 144 14.5


line stmt bran cond sub pod time code
1             # A client is a user connection for sending requests to us. Requests
2             # can either be normal user requests to be sent to a QueryWorker
3             # or management requests that start with a !.
4              
5             package MogileFS::Connection::Client;
6              
7 21     21   112 use strict;
  21         39  
  21         758  
8 21     21   67118 use Danga::Socket ();
  21         899644  
  21         819  
9 21     21   201 use base qw{Danga::Socket};
  21         47  
  21         6512  
10              
11 21     21   150 use fields qw{read_buf};
  21         51  
  21         197  
12              
13             sub new {
14 0     0 1   my $self = shift;
15 0 0         $self = fields::new($self) unless ref $self;
16 0           $self->SUPER::new( @_ );
17 0           $self->watch_read(1);
18 0           return $self;
19             }
20              
21             # Client
22             sub event_read {
23 0     0 1   my MogileFS::Connection::Client $self = shift;
24              
25 0           my $bref = $self->read(1024);
26 0 0         return $self->close unless defined $bref;
27 0           $self->{read_buf} .= $$bref;
28              
29 0           while ($self->{read_buf} =~ s/^(.*?)\r?\n//) {
30 0 0         next unless length $1;
31 0           $self->handle_request($1);
32             }
33             }
34              
35             sub handle_request {
36 0     0 0   my ($self, $line) = @_;
37              
38             # if it's just 'help', 'h', '?', or something, do that
39             #if ((substr($line, 0, 1) eq '?') || ($line eq 'help')) {
40             # MogileFS::ProcManager->SendHelp($_[1]);
41             # return;
42             #}
43              
44 0 0         if ($line =~ /^!(\S+)(?:\s+(.+))?$/) {
45 0           my ($cmd, $args) = ($1, $2);
46 0           return $self->handle_admin_command($cmd, $args);
47             }
48              
49 0           MogileFS::ProcManager->EnqueueCommandRequest($line, $self);
50             }
51              
52             sub handle_admin_command {
53 0     0 0   my ($self, $cmd, $args) = @_;
54              
55 0           my @out;
56 0 0 0       if ($cmd =~ /^stats$/) {
    0          
    0          
    0          
    0          
    0          
    0          
    0          
    0          
57             # print out some stats on the queues
58 0           my $uptime = time() - MogileFS::ProcManager->server_starttime;
59 0           my $ccount = MogileFS::ProcManager->PendingQueryCount;
60 0           my $wcount = MogileFS::ProcManager->BoredQueryWorkerCount;
61 0           my $ipcount = MogileFS::ProcManager->QueriesInProgressCount;
62 0           my $stats = MogileFS::ProcManager->StatsHash;
63 0           push @out, "uptime $uptime",
64             "pending_queries $ccount",
65             "processing_queries $ipcount",
66             "bored_queryworkers $wcount",
67 0           map { "$_ $stats->{$_}" } sort keys %$stats;
68              
69             } elsif ($cmd =~ /^shutdown/) {
70 0           print "User requested shutdown: $args\n";
71 0           kill 15, $$; # kill us, that kills our kids
72              
73             } elsif ($cmd =~ /^jobs/) {
74             # dump out a list of running jobs and pids
75             MogileFS::ProcManager->foreach_job(sub {
76 0     0     my ($job, $ct, $desired, $pidlist) = @_;
77 0           push @out, "$job count $ct";
78 0           push @out, "$job desired $desired";
79 0           push @out, "$job pids " . join(' ', @$pidlist);
80 0           });
81              
82             } elsif ($cmd =~ /^want/) {
83             # !want
84             # set the new desired staffing level for a class
85 0 0         if ($args =~ /^(\d+)\s+(\S+)/) {
86 0           my ($count, $job) = ($1, $2);
87              
88 0 0         $count = 500 if $count > 500;
89              
90             # now make sure it's a real job
91 0 0         if (MogileFS::ProcManager->is_monitor_good) {
92 0 0         if (MogileFS::ProcManager->is_valid_job($job)) {
93 0           MogileFS::ProcManager->request_job_process($job, $count);
94 0           push @out, "Now desiring $count children doing '$job'.";
95             } else {
96 0           my $classes = join(", ", MogileFS::ProcManager->valid_jobs);
97 0           push @out, "ERROR: Invalid class '$job'. Valid classes: $classes";
98             }
99             } else {
100 0           push @out, "ERROR: Monitor has not completed initial run yet\n";
101             }
102             } else {
103 0           push @out, "ERROR: usage: !want ";
104             }
105              
106             } elsif ($cmd =~ /^to/) {
107             # !to
108             # sends to all children of
109 0 0         if ($args =~ /^(\S+)\s+(.+)/) {
110 0           my $ct = MogileFS::ProcManager->ImmediateSendToChildrenByJob($1, $2);
111 0           push @out, "Message sent to $ct children.";
112              
113             } else {
114 0           push @out, "ERROR: usage: !to ";
115             }
116              
117             } elsif ($cmd =~ /^queue/ || $cmd =~ /^pend/) {
118             MogileFS::ProcManager->foreach_pending_query(sub {
119 0     0     my ($client, $query) = @_;
120 0           push @out, $query;
121 0           });
122              
123             } elsif ($cmd =~ /^watch/) {
124 0 0         if (MogileFS::ProcManager->RemoveErrorWatcher($self)) {
125 0           push @out, "Removed you from watcher list.";
126             } else {
127 0           MogileFS::ProcManager->AddErrorWatcher($self);
128 0           push @out, "Added you to watcher list.";
129             }
130              
131             } elsif ($cmd =~ /^recent/) {
132             # show the most recent N queries
133 0           push @out, MogileFS::ProcManager->RecentQueries;
134              
135             } elsif ($cmd =~ /^version/) {
136             # show the most recent N queries
137 0           push @out, $MogileFS::Server::VERSION;
138              
139             } else {
140 0           MogileFS::ProcManager->SendHelp($self, $args);
141             }
142              
143 0 0         $self->write(join("\r\n", @out) . "\r\n") if @out;
144 0           $self->write(".\r\n");
145 0           return;
146             }
147              
148             # Client
149 0     0 1   sub event_err { my $self = shift; $self->close; }
  0            
150 0     0 1   sub event_hup { my $self = shift; $self->close; }
  0            
151              
152             # just note that we've died
153             sub close {
154             # mark us as being dead
155 0     0 1   my $self = shift;
156 0           MogileFS::ProcManager->NoteDeadClient($self);
157 0           $self->SUPER::close(@_);
158             }
159              
160             1;
161              
162             # Local Variables:
163             # mode: perl
164             # c-basic-indent: 4
165             # indent-tabs-mode: nil
166             # End: