]>
git.proxmox.com Git - qemu-server.git/blob - PVE/QMPClient.pm
1 package PVE
::QMPClient
;
8 use POSIX
qw(EINTR EAGAIN);
9 use Scalar
::Util
qw(weaken);
10 use Time
::HiRes
qw(usleep gettimeofday tv_interval);
13 use PVE
::QemuServer
::Helpers
;
15 # QEMU Monitor Protocol (QMP) client.
17 # This implementation uses IO::Multiplex (libio-multiplex-perl) and
18 # allows you to issue qmp and qga commands to different VMs in parallel.
20 # Note: qemu can only handle 1 connection, so we close connections asap
23 my ($class, $eventcb) = @_;
25 my $mux = IO
::Multiplex-
>new();
29 queue_lookup
=> {}, # $fh => $queue_info
33 $self->{eventcb
} = $eventcb if $eventcb;
35 $mux->set_callback_object($self);
37 # make sure perl doesn't believe this is a circular reference as we
38 # delete mux in DESTROY
39 weaken
($mux->{_object
});
44 # Note: List of special QGA command. Those commands can close the connection
45 # without sending a response.
47 my $qga_allow_close_cmds = {
48 'guest-shutdown' => 1,
49 'guest-suspend-ram' => 1,
50 'guest-suspend-disk' => 1,
51 'guest-suspend-hybrid' => 1,
54 my $push_cmd_to_queue = sub {
55 my ($self, $vmid, $cmd) = @_;
57 my $execute = $cmd->{execute
} || die "no command name specified";
59 my $qga = ($execute =~ /^guest\-+/) ?
1 : 0;
61 my $sname = PVE
::QemuServer
::Helpers
::qmp_socket
($vmid, $qga);
63 $self->{queue_info
}->{$sname} = { qga
=> $qga, vmid
=> $vmid, sname
=> $sname, cmds
=> [] }
64 if !$self->{queue_info
}->{$sname};
66 push @{$self->{queue_info
}->{$sname}->{cmds
}}, $cmd;
68 return $self->{queue_info
}->{$sname};
71 # add a single command to the queue for later execution
72 # with queue_execute()
74 my ($self, $vmid, $callback, $execute, %params) = @_;
77 $cmd->{execute
} = $execute;
78 $cmd->{arguments
} = \
%params;
79 $cmd->{callback
} = $callback;
81 &$push_cmd_to_queue($self, $vmid, $cmd);
86 # execute a single command
88 my ($self, $vmid, $cmd, $timeout) = @_;
93 my ($vmid, $resp) = @_;
94 $result = $resp->{'return'};
95 $result = { error
=> $resp->{'error'} } if !defined($result) && $resp->{'error'};
98 die "no command specified" if !($cmd && $cmd->{execute
});
100 $cmd->{callback
} = $callback;
101 $cmd->{arguments
} = {} if !defined($cmd->{arguments
});
103 my $queue_info = &$push_cmd_to_queue($self, $vmid, $cmd);
106 # hack: monitor sometime blocks
107 if ($cmd->{execute
} eq 'query-migrate') {
108 $timeout = 60*60; # 1 hour
109 } elsif ($cmd->{execute
} =~ m/^(eject|change)/) {
110 $timeout = 60; # note: cdrom mount command is slow
111 } elsif ($cmd->{execute
} eq 'guest-fsfreeze-freeze') {
112 # freeze syncs all guest FS, if we kill it it stays in an unfreezable
113 # locked state with high probability, so use an generous timeout
114 $timeout = 60*60; # 1 hour
115 } elsif ($cmd->{execute
} eq 'guest-fsfreeze-thaw') {
116 # While it should return instantly or never (dead locked) for Linux guests,
117 # the variance for Windows guests can be big. And there might be hook scripts
118 # that are executed upon thaw, so use 3 minutes to be on the safe side.
121 $cmd->{execute
} eq 'savevm-start' ||
122 $cmd->{execute
} eq 'savevm-end' ||
123 $cmd->{execute
} eq 'query-backup' ||
124 $cmd->{execute
} eq 'query-block-jobs' ||
125 $cmd->{execute
} eq 'block-job-cancel' ||
126 $cmd->{execute
} eq 'block-job-complete' ||
127 $cmd->{execute
} eq 'backup-cancel' ||
128 $cmd->{execute
} eq 'query-savevm' ||
129 $cmd->{execute
} eq 'delete-drive-snapshot' ||
130 $cmd->{execute
} eq 'guest-fstrim' ||
131 $cmd->{execute
} eq 'guest-shutdown' ||
132 $cmd->{execute
} eq 'blockdev-snapshot-internal-sync' ||
133 $cmd->{execute
} eq 'blockdev-snapshot-delete-internal-sync' ||
134 $cmd->{execute
} eq 'snapshot-drive'
136 $timeout = 10*60; # 10 mins ?
138 # NOTE: if you came here as user and want to change this, try using IO-Threads first
139 # which move out quite some processing of the main thread, leaving more time for QMP
140 $timeout = 5; # default
144 $self->queue_execute($timeout, 2);
146 die "VM $vmid qmp command '$cmd->{execute}' failed - $queue_info->{error}"
147 if defined($queue_info->{error
});
153 my $cmdid_seq_qga = 0;
155 my $next_cmdid = sub {
160 return "$$"."0".$cmdid_seq_qga;
163 return "$$:$cmdid_seq";
167 my $lookup_queue_info = sub {
168 my ($self, $fh, $noerr) = @_;
170 my $queue_info = $self->{queue_lookup
}->{$fh};
172 warn "internal error - unable to lookup queue info" if !$noerr;
178 my $close_connection = sub {
179 my ($self, $queue_info) = @_;
181 if (my $fh = delete $queue_info->{fh
}) {
182 delete $self->{queue_lookup
}->{$fh};
183 $self->{mux
}->close($fh);
187 my $open_connection = sub {
188 my ($self, $queue_info, $timeout) = @_;
190 die "duplicate call to open" if defined($queue_info->{fh
});
192 my $vmid = $queue_info->{vmid
};
193 my $qga = $queue_info->{qga
};
195 my $sname = PVE
::QemuServer
::Helpers
::qmp_socket
($vmid, $qga);
197 $timeout = 1 if !$timeout;
200 my $starttime = [gettimeofday
];
203 my $sotype = $qga ?
'qga' : 'qmp';
207 $fh = IO
::Socket
::UNIX-
>new(Peer
=> $sname, Blocking
=> 0, Timeout
=> 1);
209 if ($! != EINTR
&& $! != EAGAIN
) {
210 die "unable to connect to VM $vmid $sotype socket - $!\n";
212 my $elapsed = tv_interval
($starttime, [gettimeofday
]);
213 if ($elapsed >= $timeout) {
214 die "unable to connect to VM $vmid $sotype socket - timeout after $count retries\n";
219 $queue_info->{fh
} = $fh;
221 $self->{queue_lookup
}->{$fh} = $queue_info;
223 $self->{mux
}->add($fh);
224 $self->{mux
}->set_timeout($fh, $timeout);
229 my $check_queue = sub {
234 foreach my $sname (keys %{$self->{queue_info
}}) {
235 my $queue_info = $self->{queue_info
}->{$sname};
236 my $fh = $queue_info->{fh
};
239 my $qga = $queue_info->{qga
};
241 if ($queue_info->{error
}) {
242 &$close_connection($self, $queue_info);
246 if ($queue_info->{current
}) { # command running, waiting for response
251 if (!scalar(@{$queue_info->{cmds
}})) { # no more commands
252 &$close_connection($self, $queue_info);
258 my $cmd = $queue_info->{current
} = shift @{$queue_info->{cmds
}};
259 $cmd->{id
} = &$next_cmdid($qga);
262 if ($cmd->{execute
} eq 'add-fd' || $cmd->{execute
} eq 'getfd') {
263 $fd = $cmd->{arguments
}->{fd
};
264 delete $cmd->{arguments
}->{fd
};
271 $qmpcmd = to_json
({ execute
=> 'guest-sync-delimited',
272 arguments
=> { id
=> int($cmd->{id
})}}) . "\n" .
273 to_json
({ execute
=> $cmd->{execute
}, arguments
=> $cmd->{arguments
}}) . "\n";
278 execute
=> $cmd->{execute
},
279 arguments
=> $cmd->{arguments
},
284 my $ret = PVE
::IPCC
::sendfd
(fileno($fh), $fd, $qmpcmd);
285 die "sendfd failed" if $ret < 0;
287 $self->{mux
}->write($fh, $qmpcmd);
291 $queue_info->{error
} = $err;
297 $self->{mux
}->endloop() if !$running;
302 # execute all queued command
305 my ($self, $timeout, $noerr) = @_;
307 $timeout = 3 if !$timeout;
309 # open all necessary connections
310 foreach my $sname (keys %{$self->{queue_info
}}) {
311 my $queue_info = $self->{queue_info
}->{$sname};
312 next if !scalar(@{$queue_info->{cmds
}}); # no commands
314 $queue_info->{error
} = undef;
315 $queue_info->{current
} = undef;
318 &$open_connection($self, $queue_info, $timeout);
320 if (!$queue_info->{qga
}) {
321 my $cap_cmd = { execute
=> 'qmp_capabilities', arguments
=> {} };
322 unshift @{$queue_info->{cmds
}}, $cap_cmd;
326 $queue_info->{error
} = $err;
334 $running = &$check_queue($self);
341 # make sure we close everything
343 foreach my $sname (keys %{$self->{queue_info
}}) {
344 my $queue_info = $self->{queue_info
}->{$sname};
345 &$close_connection($self, $queue_info);
346 if ($queue_info->{error
}) {
348 warn $queue_info->{error
} if $noerr < 2;
350 $errors .= $queue_info->{error
}
355 $self->{queue_info
} = $self->{queue_lookup
} = {};
357 die $errors if $errors;
361 my ($self, $mux, $fh) = @_;
363 my $queue_info = &$lookup_queue_info($self, $fh, 1);
364 return if !$queue_info;
366 $queue_info->{error
} = "client closed connection\n"
367 if !$queue_info->{error
};
370 # mux_input is called when input is available on one of the descriptors.
372 my ($self, $mux, $fh, $input) = @_;
374 my $queue_info = &$lookup_queue_info($self, $fh);
375 return if !$queue_info;
377 my $sname = $queue_info->{sname
};
378 my $vmid = $queue_info->{vmid
};
379 my $qga = $queue_info->{qga
};
381 my $curcmd = $queue_info->{current
};
382 die "unable to lookup current command for VM $vmid ($sname)\n" if !$curcmd;
387 return if $$input !~ s/^.*\xff([^\n]+}\r?\n[^\n]+})\r?\n(.*)$/$2/so;
390 return if $$input !~ s/^(.*})\r?\n(.*)$/$2/so;
395 my @jsons = split("\n", $raw);
399 die "response is not complete" if @jsons != 2 ;
401 my $obj = from_json
($jsons[0]);
403 my $cmdid = $obj->{'return'};
404 die "received responsed without command id\n" if !$cmdid;
406 # skip results fro previous commands
407 return if $cmdid < $curcmd->{id
};
409 if ($curcmd->{id
} ne $cmdid) {
410 die "got wrong command id '$cmdid' (expected $curcmd->{id})\n";
413 delete $queue_info->{current
};
415 $obj = from_json
($jsons[1]);
417 if (my $callback = $curcmd->{callback
}) {
418 &$callback($vmid, $obj);
424 foreach my $json (@jsons) {
425 my $obj = from_json
($json);
426 next if defined($obj->{QMP
}); # skip monitor greeting
428 if (exists($obj->{error
}->{desc
})) {
429 my $desc = $obj->{error
}->{desc
};
431 die "$desc\n" if $desc !~ m/Connection can not be completed immediately/;
435 if (defined($obj->{event
})) {
436 if (my $eventcb = $self->{eventcb
}) {
442 my $cmdid = $obj->{id
};
443 die "received responsed without command id\n" if !$cmdid;
445 if ($curcmd->{id
} ne $cmdid) {
446 die "got wrong command id '$cmdid' (expected $curcmd->{id})\n";
449 delete $queue_info->{current
};
451 if (my $callback = $curcmd->{callback
}) {
452 &$callback($vmid, $obj);
457 $queue_info->{error
} = $err;
460 &$check_queue($self);
463 # This gets called every second to update player info, etc...
465 my ($self, $mux, $fh) = @_;
467 if (my $queue_info = &$lookup_queue_info($self, $fh)) {
468 $queue_info->{error
} = "got timeout\n";
469 $self->{mux
}->inbuffer($fh, ''); # clear to avoid warnings
472 &$check_queue($self);
476 my ($self, $mux, $fh, $input) = @_;
478 my $queue_info = &$lookup_queue_info($self, $fh);
479 return if !$queue_info;
481 my $sname = $queue_info->{sname
};
482 my $vmid = $queue_info->{vmid
};
483 my $qga = $queue_info->{qga
};
485 my $curcmd = $queue_info->{current
};
486 die "unable to lookup current command for VM $vmid ($sname)\n" if !$curcmd;
488 if ($qga && $qga_allow_close_cmds->{$curcmd->{execute
}}) {
490 return if $$input !~ s/^.*\xff([^\n]+})\r?\n(.*)$/$2/so;
495 my $obj = from_json
($raw);
497 my $cmdid = $obj->{'return'};
498 die "received responsed without command id\n" if !$cmdid;
500 delete $queue_info->{current
};
502 if (my $callback = $curcmd->{callback
}) {
503 &$callback($vmid, undef);
507 $queue_info->{error
} = $err;
510 &$close_connection($self, $queue_info);
512 if (scalar(@{$queue_info->{cmds
}}) && !$queue_info->{error
}) {
513 $queue_info->{error
} = "Got EOF but command queue is not empty.\n";
521 foreach my $sname (keys %{$self->{queue_info
}}) {
522 my $queue_info = $self->{queue_info
}->{$sname};
523 $close_connection->($self, $queue_info);