1. my class Proc::Async { ... }
  2. my role X::Proc::Async is Exception {
  3. has Proc::Async $.proc;
  4. }
  5. my class X::Proc::Async::TapBeforeSpawn does X::Proc::Async {
  6. has $.handle;
  7. method message() {
  8. "To avoid data races, you must tap $!handle before running the process"
  9. }
  10. }
  11. my class X::Proc::Async::SupplyOrStd does X::Proc::Async {
  12. method message() {
  13. "Using .Supply on a Proc::Async implies merging stdout and stderr; .stdout " ~
  14. "and .stderr cannot therefore be used in combination with it"
  15. }
  16. }
  17. my class X::Proc::Async::BindOrUse does X::Proc::Async {
  18. has $.handle;
  19. has $.use;
  20. method message() {
  21. "Cannot both bind $.handle to a handle and also $.use"
  22. }
  23. }
  24. my class X::Proc::Async::CharsOrBytes does X::Proc::Async {
  25. has $.handle;
  26. method message() {
  27. "Can only tap one of chars or bytes supply for $!handle"
  28. }
  29. }
  30. my class X::Proc::Async::AlreadyStarted does X::Proc::Async {
  31. method message() {
  32. "Process has already been started"
  33. }
  34. }
  35. my class X::Proc::Async::MustBeStarted does X::Proc::Async {
  36. has $.method;
  37. method message() {
  38. "Process must be started first before calling '$!method'"
  39. }
  40. }
  41. my class X::Proc::Async::OpenForWriting does X::Proc::Async {
  42. has $.method;
  43. method message() {
  44. "Process must be opened for writing with :w to call '$!method'"
  45. }
  46. }
  47. my class Proc::Async {
  48. # An asynchornous process output pipe is a Supply that also can provide
  49. # the native descriptor of the underlying pipe.
  50. class Pipe is Supply {
  51. my class PermitOnTap does Tappable {
  52. has Tappable $.delegate;
  53. has &.on-tap;
  54. method tap(|c) {
  55. &!on-tap();
  56. $!delegate.tap(|c)
  57. }
  58. method live() { self.delegate.live }
  59. method serial() { self.delegate.serial }
  60. method sane() { self.delegate.sane }
  61. }
  62. has Promise $.native-descriptor;
  63. has &!on-nd-used;
  64. submethod BUILD(:$!native-descriptor!, :&!on-nd-used) {}
  65. method native-descriptor() {
  66. &!on-nd-used();
  67. $!native-descriptor
  68. }
  69. method new($delegate, $native-descriptor, &on-tap, &on-nd-used) {
  70. self.bless(
  71. tappable => PermitOnTap.bless(:$delegate, :&on-tap),
  72. :$native-descriptor, :&on-nd-used
  73. )
  74. }
  75. }
  76. my class ProcessCancellation is repr('AsyncTask') { }
  77. my enum CharsOrBytes ( :Bytes(0), :Chars(1) );
  78. has $!ready_promise = Promise.new;
  79. has $!ready_vow = $!ready_promise.vow;
  80. has $!handle_available_promise = Promise.new;
  81. has $!stdout_descriptor_vow;
  82. has $!stderr_descriptor_vow;
  83. has $!stdout_descriptor_used = Promise.new;
  84. has $!stderr_descriptor_used = Promise.new;
  85. has $.path;
  86. has @.args;
  87. has $.w;
  88. has $.enc = 'utf8';
  89. has $.translate-nl = True;
  90. has Bool $.started = False;
  91. has $!stdout_supply;
  92. has CharsOrBytes $!stdout_type;
  93. has $!stderr_supply;
  94. has CharsOrBytes $!stderr_type;
  95. has $!merge_supply;
  96. has CharsOrBytes $!merge_type;
  97. has $!stdin-fd;
  98. has $!stdout-fd;
  99. has $!stderr-fd;
  100. has $!process_handle;
  101. has $!exit_promise;
  102. has @!promises;
  103. has $!encoder;
  104. has @!close-after-exit;
  105. proto method new(|) {*}
  106. multi method new(*@args where .so) {
  107. my $path = @args.shift;
  108. self.bless(:$path, :@args, |%_)
  109. }
  110. submethod TWEAK(--> Nil) {
  111. $!encoder := Encoding::Registry.find($!enc).encoder(:$!translate-nl);
  112. }
  113. method !pipe-cbs(\channel) {
  114. -> { $!handle_available_promise.then({ nqp::permit($!process_handle, channel, -1) }) },
  115. -> { (channel == 1 ?? $!stdout_descriptor_used !! $!stderr_descriptor_used).keep(True) }
  116. }
  117. method !pipe(\what, \the-supply, \type, \value, \fd-vow, \permit-channel) {
  118. X::Proc::Async::TapBeforeSpawn.new(handle => what, proc => self).throw
  119. if $!started;
  120. X::Proc::Async::CharsOrBytes.new(handle => what, proc => self).throw
  121. if the-supply and type != value;
  122. type = value;
  123. the-supply //= Supplier::Preserving.new;
  124. if nqp::iscont(fd-vow) {
  125. my $native-descriptor = Promise.new;
  126. fd-vow = $native-descriptor.vow;
  127. Pipe.new(the-supply.Supply.Tappable, $native-descriptor, |self!pipe-cbs(permit-channel))
  128. }
  129. else {
  130. the-supply.Supply
  131. }
  132. }
  133. method !wrap-decoder(Supply:D $bin-supply, $enc, \fd-vow, \permit-channel, :$translate-nl) {
  134. my \sup = Rakudo::Internals.BYTE_SUPPLY_DECODER($bin-supply, $enc // $!enc,
  135. :translate-nl($translate-nl // $!translate-nl));
  136. if nqp::iscont(fd-vow) {
  137. my $native-descriptor = Promise.new;
  138. fd-vow = $native-descriptor.vow;
  139. Pipe.new(sup.Supply.Tappable, $native-descriptor, |self!pipe-cbs(permit-channel))
  140. }
  141. else {
  142. sup
  143. }
  144. }
  145. proto method stdout(|) {*}
  146. multi method stdout(Proc::Async:D: :$bin!) {
  147. die X::Proc::Async::SupplyOrStd.new if $!merge_supply;
  148. die X::Proc::Async::BindOrUse.new(:handle<stdout>, :use('get the stdout Supply'))
  149. if $!stdout-fd;
  150. $bin
  151. ?? self!pipe('stdout', $!stdout_supply, $!stdout_type, Bytes, $!stdout_descriptor_vow, 1)
  152. !! self.stdout(|%_)
  153. }
  154. multi method stdout(Proc::Async:D: :$enc, :$translate-nl) {
  155. die X::Proc::Async::SupplyOrStd.new if $!merge_supply;
  156. die X::Proc::Async::BindOrUse.new(:handle<stdout>, :use('get the stdout Supply'))
  157. if $!stdout-fd;
  158. self!wrap-decoder:
  159. self!pipe('stdout', $!stdout_supply, $!stdout_type, Chars, Nil, 1),
  160. $enc, $!stdout_descriptor_vow, 1, :$translate-nl
  161. }
  162. proto method stderr(|) {*}
  163. multi method stderr(Proc::Async:D: :$bin!) {
  164. die X::Proc::Async::SupplyOrStd.new if $!merge_supply;
  165. die X::Proc::Async::BindOrUse.new(:handle<stderr>, :use('get the stderr Supply'))
  166. if $!stderr-fd;
  167. $bin
  168. ?? self!pipe('stderr', $!stderr_supply, $!stderr_type, Bytes, $!stderr_descriptor_vow, 2)
  169. !! self.stderr(|%_)
  170. }
  171. multi method stderr(Proc::Async:D: :$enc, :$translate-nl) {
  172. die X::Proc::Async::SupplyOrStd.new if $!merge_supply;
  173. die X::Proc::Async::BindOrUse.new(:handle<stderr>, :use('get the stderr Supply'))
  174. if $!stderr-fd;
  175. self!wrap-decoder:
  176. self!pipe('stderr', $!stderr_supply, $!stderr_type, Chars, Nil, 2),
  177. $enc, $!stderr_descriptor_vow, 2, :$translate-nl
  178. }
  179. proto method Supply(|) {*}
  180. multi method Supply(Proc::Async:D: :$bin!) {
  181. die X::Proc::Async::SupplyOrStd.new if $!stdout_supply || $!stderr_supply;
  182. die X::Proc::Async::BindOrUse.new(:handle<stdout>, :use('get the output Supply'))
  183. if $!stdout-fd;
  184. die X::Proc::Async::BindOrUse.new(:handle<stderr>, :use('get the output Supply'))
  185. if $!stderr-fd;
  186. $bin
  187. ?? self!pipe('merge', $!merge_supply, $!merge_type, Bytes, Nil, 0)
  188. !! self.Supply(|%_)
  189. }
  190. multi method Supply(Proc::Async:D: :$enc, :$translate-nl) {
  191. die X::Proc::Async::SupplyOrStd.new if $!stdout_supply || $!stderr_supply;
  192. die X::Proc::Async::BindOrUse.new(:handle<stdout>, :use('get the output Supply'))
  193. if $!stdout-fd;
  194. die X::Proc::Async::BindOrUse.new(:handle<stderr>, :use('get the output Supply'))
  195. if $!stderr-fd;
  196. self!wrap-decoder:
  197. self!pipe('merge', $!merge_supply, $!merge_type, Chars, Nil, 0),
  198. $enc, Nil, 0, :$translate-nl
  199. }
  200. proto method bind-stdin($) {*}
  201. multi method bind-stdin(IO::Handle:D $handle --> Nil) {
  202. die X::Proc::Async::BindOrUse.new(:handle<stdin>, :use('use :w')) if $!w;
  203. $!stdin-fd := $handle.native-descriptor;
  204. @!close-after-exit.push($handle) if $handle ~~ IO::Pipe;
  205. }
  206. multi method bind-stdin(Proc::Async::Pipe:D $pipe --> Nil) {
  207. die X::Proc::Async::BindOrUse.new(:handle<stdin>, :use('use :w')) if $!w;
  208. $!stdin-fd := $pipe.native-descriptor;
  209. }
  210. method bind-stdout(IO::Handle:D $handle --> Nil) {
  211. die X::Proc::Async::BindOrUse.new(:handle<stdout>, :use('get the stdout Supply'))
  212. if $!stdout_supply;
  213. die X::Proc::Async::BindOrUse.new(:handle<stdout>, :use('get the output Supply'))
  214. if $!merge_supply;
  215. $!stdout-fd := $handle.native-descriptor;
  216. }
  217. method bind-stderr(IO::Handle:D $handle --> Nil) {
  218. die X::Proc::Async::BindOrUse.new(:handle<stderr>, :use('get the stderr Supply'))
  219. if $!stderr_supply;
  220. die X::Proc::Async::BindOrUse.new(:handle<stderr>, :use('get the output Supply'))
  221. if $!merge_supply;
  222. $!stderr-fd := $handle.native-descriptor;
  223. }
  224. method ready(--> Promise) {
  225. $!ready_promise;
  226. }
  227. method !capture(\callbacks,\std,\the-supply) {
  228. my $promise = Promise.new;
  229. my $vow = $promise.vow;
  230. my $ss = Rakudo::Internals::SupplySequencer.new(
  231. on-data-ready => -> \data { the-supply.emit(data) },
  232. on-completed => -> { the-supply.done(); $vow.keep(the-supply) },
  233. on-error => -> \err { the-supply.quit(err); $vow.keep((the-supply,err)) });
  234. nqp::bindkey(callbacks,
  235. std ~ '_bytes' ,
  236. -> Mu \seq, Mu \data, Mu \err { $ss.process(seq, data, err) });
  237. $promise;
  238. }
  239. method start(Proc::Async:D: :$scheduler = $*SCHEDULER, :$ENV, :$cwd = $*CWD) {
  240. X::Proc::Async::AlreadyStarted.new(proc => self).throw if $!started;
  241. $!started = True;
  242. my @blockers;
  243. if $!stdin-fd ~~ Promise {
  244. @blockers.push($!stdin-fd.then({ $!stdin-fd := .result }));
  245. }
  246. @blockers
  247. ?? start { await @blockers; await self!start-internal(:$scheduler, :$ENV, :$cwd) }
  248. !! self!start-internal(:$scheduler, :$ENV, :$cwd)
  249. }
  250. method !start-internal(:$scheduler, :$ENV, :$cwd) {
  251. my %ENV := $ENV ?? $ENV.hash !! %*ENV;
  252. $!exit_promise = Promise.new;
  253. my Mu $callbacks := nqp::hash();
  254. nqp::bindkey($callbacks, 'done', -> Mu \status {
  255. $!exit_promise.keep(Proc.new(
  256. :exitcode(status +> 8), :signal(status +& 0xFF),
  257. :command[ $!path, |@!args ],
  258. ))
  259. });
  260. nqp::bindkey($callbacks, 'ready', -> Mu \handles = Nil {
  261. if nqp::isconcrete(handles) {
  262. with $!stdout_descriptor_vow {
  263. my $fd = nqp::atpos_i(handles, 0);
  264. $fd < 0
  265. ?? .break("Descriptor not available")
  266. !! .keep($fd)
  267. }
  268. with $!stderr_descriptor_vow {
  269. my $fd = nqp::atpos_i(handles, 1);
  270. $fd < 0
  271. ?? .break("Descriptor not available")
  272. !! .keep($fd)
  273. }
  274. }
  275. $!ready_vow.keep(Nil);
  276. });
  277. nqp::bindkey($callbacks, 'error', -> Mu \err {
  278. my $error = X::OS.new(os-error => err);
  279. $!exit_promise.break($error);
  280. $!ready_vow.break($error);
  281. });
  282. @!promises.push(Promise.anyof(
  283. self!capture($callbacks,'stdout',$!stdout_supply),
  284. $!stdout_descriptor_used
  285. )) if $!stdout_supply;
  286. @!promises.push(Promise.anyof(
  287. self!capture($callbacks,'stderr',$!stderr_supply),
  288. $!stderr_descriptor_used
  289. )) if $!stderr_supply;
  290. @!promises.push(
  291. self!capture($callbacks,'merge',$!merge_supply)
  292. ) if $!merge_supply;
  293. nqp::bindkey($callbacks, 'buf_type', buf8.new);
  294. nqp::bindkey($callbacks, 'write', True) if $.w;
  295. nqp::bindkey($callbacks, 'stdin_fd', $!stdin-fd) if $!stdin-fd.DEFINITE;
  296. nqp::bindkey($callbacks, 'stdout_fd', $!stdout-fd) if $!stdout-fd.DEFINITE;
  297. nqp::bindkey($callbacks, 'stderr_fd', $!stderr-fd) if $!stderr-fd.DEFINITE;
  298. $!process_handle := nqp::spawnprocasync($scheduler.queue(:hint-affinity),
  299. CLONE-LIST-DECONTAINERIZED($!path,@!args),
  300. $cwd.Str,
  301. CLONE-HASH-DECONTAINERIZED(%ENV),
  302. $callbacks,
  303. );
  304. $!handle_available_promise.keep(True);
  305. nqp::permit($!process_handle, 0, -1) if $!merge_supply;
  306. Promise.allof( $!exit_promise, @!promises ).then({
  307. .close for @!close-after-exit;
  308. $!exit_promise.status == Broken
  309. ?? $!exit_promise.cause.throw
  310. !! $!exit_promise.result
  311. })
  312. }
  313. method print(Proc::Async:D: Str() $str, :$scheduler = $*SCHEDULER) {
  314. X::Proc::Async::OpenForWriting.new(:method<print>, proc => self).throw if !$!w;
  315. X::Proc::Async::MustBeStarted.new(:method<print>, proc => self).throw if !$!started;
  316. self.write($!encoder.encode-chars($str))
  317. }
  318. method put(Proc::Async:D: \x, |c) {
  319. X::Proc::Async::OpenForWriting.new(:method<say>, proc => self).throw if !$!w;
  320. X::Proc::Async::MustBeStarted.new(:method<say>, proc => self).throw if !$!started;
  321. self.print( x.join ~ "\n", |c );
  322. }
  323. method say(Proc::Async:D: \x, |c) {
  324. X::Proc::Async::OpenForWriting.new(:method<say>, proc => self).throw if !$!w;
  325. X::Proc::Async::MustBeStarted.new(:method<say>, proc => self).throw if !$!started;
  326. self.print( x.gist ~ "\n", |c );
  327. }
  328. method write(Proc::Async:D: Blob:D $b, :$scheduler = $*SCHEDULER) {
  329. X::Proc::Async::OpenForWriting.new(:method<write>, proc => self).throw if !$!w;
  330. X::Proc::Async::MustBeStarted.new(:method<write>, proc => self).throw if !$!started;
  331. my $p = Promise.new;
  332. my $v = $p.vow;
  333. nqp::asyncwritebytes(
  334. $!process_handle,
  335. $scheduler.queue,
  336. -> Mu \bytes, Mu \err {
  337. if err {
  338. $v.break(err);
  339. }
  340. else {
  341. $v.keep(bytes);
  342. }
  343. },
  344. nqp::decont($b), ProcessCancellation);
  345. $p
  346. }
  347. method close-stdin(Proc::Async:D:) {
  348. X::Proc::Async::OpenForWriting.new(:method<close-stdin>, proc => self).throw
  349. if !$!w;
  350. X::Proc::Async::MustBeStarted.new(:method<close-stdin>, proc => self).throw
  351. if !$!started;
  352. nqp::closefh($!process_handle);
  353. True;
  354. }
  355. # Note: some of the duplicated code in methods could be moved to
  356. # proto, but at the moment (2017-06-02) that makes the call 24% slower
  357. proto method kill(|) {*}
  358. multi method kill(Proc::Async:D: Signal:D \signal = SIGHUP) {
  359. X::Proc::Async::MustBeStarted.new(:method<kill>, proc => self).throw if !$!started;
  360. nqp::killprocasync($!process_handle, signal.value)
  361. }
  362. multi method kill(Proc::Async:D: Int:D \signal) {
  363. X::Proc::Async::MustBeStarted.new(:method<kill>, proc => self).throw if !$!started;
  364. nqp::killprocasync($!process_handle, signal)
  365. }
  366. multi method kill(Proc::Async:D: Str:D \signal) {
  367. X::Proc::Async::MustBeStarted.new(:method<kill>, proc => self).throw if !$!started;
  368. nqp::killprocasync($!process_handle, $*KERNEL.signal: signal)
  369. }
  370. }