Playing with Go: Embarrassingly Parallel Scripts

http://collectiveidea.com/blog/archives/2012/12/03/playing-with-go-embarrassingly-parallel-scripts/

by jameskilton • 14 years ago

97 52 14 years ago

jgrahamc• 14 years ago

Yes, Go does make writing stuff like that nice, but I think the actual code given is pretty heavyweight. This does the same thing:

  package main

  import (
    "fmt"
    "net"
    "io/ioutil"
    "strings"
  )

  func main() {
    file_in, _ := ioutil.ReadFile("domains.txt")
    domain_list := string(file_in)

    done := make(chan bool)
    count := 0

    for _, domain := range strings.Split(strings.TrimSpace(domain_list), "\n") {
      go func(d string) {
        ipAddresses, _ := net.LookupIP(d)

        ip := ""
        if len(ipAddresses) > 0 {
          ip = ipAddresses[0].String()
        }

        fmt.Println("Mapping: ", d, "->", ip)
        done <- true
      }(domain)

      count++
    }

    for i := 0; i < count; i++ {
  	<- done
    }
  }

dadkins• 14 years ago

There's a library for that pattern: http://golang.org/pkg/sync/#WaitGroup

Instead of counting and using a done channel,

    import "sync"
    ...
    var wg sync.WaitGroup
    for ... {
        wg.Add(1)
        go dowork()
    }
    wg.Wait()

jameskilton2• 14 years ago

Nice. Yeah I'm very new to Go so I know almost nothing about the stdlib libraries. Thanks for the pointer.

zmj• 14 years ago

I'm embarrassed how many times I've implemented the above without thinking to look for a standard lib solution. Thanks!

0xABADC0DA2• 14 years ago

Even so why would you write that boilerplate code out each time?

Something like this would work even better:

    result = src.asyncMap { |e| dowork(e) };

Except Google Go returns several values instead of tuples, so you can't just collect all the results as-is. And with no generics it would be annoying to actually use e and the result, since they would need casts. Too bad.

y0ghur7_xxx• 14 years ago

I know it's about Go, but bash is awesome for stuff like this:

   while read line; do
   host $line | head -n1 | awk '{print $1 " -> " $4}' &
   done < domains.txt

etrain2• 14 years ago

I realize it's a one-off, but you could potentially mix lines of output with a script like that.

If you gnu parallel (xargs++) installed - turn that second line into a script (and replace $line with $1).

  cat domains.txt | parallel -n 1 -P 50 script.sh

philsnow• 14 years ago

You don't need to spawn N awks,

  <domains.txt while read line; do
    host $line | head -n1 &
  done | awk '{print $1 " -> " $4}'

drivebyacct2• 14 years ago

So many ignored errors. :(

jameskilton2• 14 years ago

It's a one-off script, I didn't really care about errors. If this was something run regularly inside of a bigger application yes I'd have full error handling.

stusmall2• 14 years ago

This is one of the things I love about go. If he wants the return value but doesn't want to address the errors, he has to actively do it. It makes you think twice when you are putting a _ for an error return. And now to another programmer coming in to maintain it, they stick out like a sore thumb.

hannibalhorn2• 14 years ago

Go is great for stuff like this, especially when part of an actual system.

That said, if this was just an adhoc job (to figure out which domains point to a specific IP address) you can just use "xargs -P" or GNU parallel and it becomes a pretty basic shell script, along the lines of: cat domains.txt | xargs -P 1000 -n 1 host

mistermcgruff• 14 years ago

xargs -P is a godsend for one-off jobs needing parallelism

skeltoac2• 14 years ago

Until last week I didn't know that xargs could invoke commands in parallel.

  xargs -n1 -P8 dig A <hosts.txt | grep -v ';' | grep $TARGET_IP | sed 's/\.\s.*//' >hosts-matched.txt

scurry• 14 years ago

So what's the difference between xargs and parallel? I thought the point of parallel was that it was xargs with the addition of running things in parallel. But if xargs can do that already, is there any reason to use one over the other?

lazyjones• 14 years ago

To be fair to languages without such great parallelism support: you can do this using asynchronous/event-loop-based code because the parallelism will be limited by the nameserver anyway (the calling code does almost nothing, it mostly waits for the net / the nameserver).

chimeracoder• 14 years ago

> To be fair to languages without such great parallelism support: you can do this using asynchronous/event-loop-based code

Well you can do with callbacks anything that you can do with channels and goroutines. Go's primary appeal is that it makes concurrent[1] code easy to reason about, not that it enables you to do anything that you "couldn't do" otherwise.

Continuations are just GOTOs, and just like GOTOs, some people love them and some people hate them, but even people who like them can find them difficult in large doses. Goroutines and channels are nice, because they fit the structure of imperative code, whereas callbacks sort of resemble imperative code but "inside out".

[1] Note that I didn't say parallel!

j2labs• 14 years ago

This kind of a thing is also very easy in Python if you use Gevent.

I have used Gevent a lot over the last couple years and see a lot of similarities in Go's concurrency, which I think is great.

Concurrency doesn't have to be about insane looking code!

d0mine• 14 years ago

With concurrent.futures http://docs.python.org/3/library/concurrent.futures.html#con...

it could look like:

  from concurrent.futures import ThreadPoolExecutor as Pool
  from socket import getaddrinfo

  def lookup(domain):
        try:
            result = getaddrinfo(domain, 80)
        exception Exception as e:
            print("error %s -> %s" % (domain, e))
        else:
            print("done  %s -> %s" % (domain, result))

  nconcurrent = 20
  with open('domains.txt') as file, Pool(nconcurrent) as pool:
        for domain in (line.strip() for line in file):
            pool.submit(lookup, domain)

To run multiple processes instead of threads, change the import to ProcessPoolExecutor.

To support multiprocessing.Pool (for Python 2 where concurrent.futures is not in stdlib), replace pool.submit() with pool.apply_async() and use contextlib.closing() around the Pool().

jnazario2• 14 years ago

in the DNS case asynchronous event handling would be super easy to do. in python asyncore with something like dpkt to construct and read DNS lookups works like a champ, as does twisted. i did a simple async DNS resolver in pure python (asyncore, dpkt) and can sustain thousands of lookups a second. GNU adns also has bindings in various languages.

you can get Go's parallelisms via CSP (e.g. python-csp, ruby-csp) and replace a lot of fragile threading/parallel code with it. i've been doing that in lieu of learning Go (i know i know .. i'm lazy) and been very pleased.

anyhow, many ways to skin cats. those are just two or three.

tptacek2• 14 years ago

You wouldn't do DNS lookups asynchronously in Go to begin with. Modeling concurrency of any sort in Go the way you would with an event loop is usually a code smell.

jnazario2• 14 years ago

agreed, and i should have been more clear. the author's blog post states that one of the reasons he explored Go was that his initial sketch of a solution in his language of choice (ruby) was sequential. my point was that you can get performance in ruby with asynchronous operations, and that you don't need to go parallel for something like this.

then again if it was a matter of "well, i had a problem to solve and i had a desire to explore another language, so solving it in that new language was a way to explore" then the point is moot.

however agreed 100% or more on the "code smell" of doing an event loop in Go.

blablabla123• 14 years ago

Not sure if I understand your point, can you explain a bit more please? Are you saying event loops with select {} are considered code smell?

dkhenry• 14 years ago

Scala is really great for this too. The downside is the spin up time for the JVM, but the upside is that if you use SBT's script launcher it will compile and cache the script transparently for you , you can still pull in _any_ JVM dependency and you can run it just like a shell script. I needed to test for a port being open in parallel and it was a cake walk to use NIO's SocketSelector to do it reactivly.

danneu• 14 years ago

My attempt with Ruby and Celluloid.

https://gist.github.com/a803d86234e8d1fc5496

I also include a list of 100 domains in a domains.txt if anyone wants to try for themselves.

    require "socket"
    require "celluloid"

    class IPGetter
      include Celluloid

      def get(url)
        Socket.getaddrinfo(url, "http")[0][2]
      end
    end

    pool = IPGetter.pool(size: 100)
    ips = {}

    File.open("domains.txt").each_line do |line|
      line.chomp!
      ips[line] = pool.future.get(line)
    end

    ips.each do |url, ip_future|
      puts "#{url} => #{ip_future.value}"
    end

ericmoritz• 14 years ago

Or Erlang

https://gist.github.com/bb01a85404e6b445dcb3#file_resolve_do...

    % -*- erlang -*-                                                                                                                              
    %%! -smp enable                                                                                                                               
                                                                                                                                                  
    worker(Hostname) ->                                                                                                                           
        {ok, IP} = inet:getaddr(Hostname, inet),                                                                                                  
        io:format(                                                                                                                                
          "~s => ~s~n",                                                                                                                                 
          [ip_to_string(IP)]                                                                                                                      
         ).                                                                                                                                       
                                                                                                                                                  
    ip_to_string({N1,N2,N3,N4}) ->                                                                                                                
        io_lib:format(                                                                                                                            
          "~w.~w.~w.~w",                                                                                                                          
          [N1,N2,N3,N4]                                                                                                                           
         ).                                                                                                                                       
                                                                                                                                                  
    main([DomainFile]) ->                                                                                                                         
        {ok, Bin} = file:read_file(DomainFile),                                                                                                   
        String = binary_to_list(Bin),                                                                                                             
        Domains = string:tokens(String, "\n"),                                                                                                    
        plists:foreach(                                                                                                                           
          fun(Domain) -> worker(Domain) end,                                                                                                      
          Domains                                                                                                                                 
        ).

This uses https://github.com/eveel/plists/

dxbydt• 14 years ago

In Scala using the par function: https://gist.github.com/4199415

    def ip(host:String)=java.net.InetAddress.getAllByName(host)(0).getHostAddress
    var map = Map[String,String]()
    val hosts = List("google.com","twitter.com","facebook.com")
    hosts.par.map(host => (host,ip(host))).foreach(hostip => map+=(hostip._1->hostip._2))

    scala> map
    res1: scala.collection.immutable.Map[String,String]
     =Map(google.com -> 74.125.224.71,
          twitter.com -> 199.59.150.7, 
          facebook.com -> 69.171.229.16)

jonhohle• 14 years ago

How about a few lines of C?

  // clang -lcares -I/opt/local/include -L/opt/local/lib -o domains domains.c
  #include <ares.h>
  #include <stdio.h>
  #include <stdlib.h>
  #include <sys/socket.h>
  #include <arpa/inet.h>
  #include <netinet/in.h>
  #include <netdb.h>
  #include <stdarg.h>
  #include <string.h>
  #include <ctype.h>
  #include <unistd.h>
  #include <sys/queue.h>
  #include <fcntl.h>
  
  static void callback(void *arg, int status, int timeouts, struct hostent *host)
  {
  
      if(!host || status != ARES_SUCCESS){
          printf("Failed to lookup %s\n", ares_strerror(status));
          return;
      }
  
      char ip[INET6_ADDRSTRLEN];
      int i = 0;
  
      for (i = 0; host->h_addr_list[i]; ++i) {
          inet_ntop(host->h_addrtype, host->h_addr_list[i], ip, sizeof(ip));
          printf("%s: %s\n", host->h_name, ip);
      }
  }
  
  static void wait_ares(ares_channel channel)
  {
      for(;;){
          struct timeval *tvp, tv;
          fd_set read_fds, write_fds;
          int nfds;
  
          FD_ZERO(&read_fds);
          FD_ZERO(&write_fds);
          nfds = ares_fds(channel, &read_fds, &write_fds);
          if(nfds == 0){
              break;
          }
          tvp = ares_timeout(channel, NULL, &tv);
          select(nfds, &read_fds, &write_fds, NULL, tvp);
          ares_process(channel, &read_fds, &write_fds);
      }
  }
  
  struct Request {
      ares_channel channel;
      char* domain;
      SLIST_ENTRY(Request) next;
  };
  
  int main(const int argc, const char* argv[]) {
      SLIST_HEAD(Requests, Request) requests;
      SLIST_INIT(&requests);
      struct Request* last = NULL;
  
      int status;
      struct ares_options options;
      int optmask = 0;
      const char* path;
  
      optmask |= ARES_OPT_TIMEOUTMS;
      options.timeout = 1000;
  
      status = ares_library_init(ARES_LIB_INIT_ALL);
      if (status != ARES_SUCCESS){
          printf("ares_library_init: %s\n", ares_strerror(status));
          return 1;
      }
  
      for (int i = 1; (path = argv[i]) != NULL || i == 0; i++) {
          
          int fd = open(path, O_RDONLY);
          FILE* fp = fdopen(fd, "r");
          char* buffer = NULL;
          size_t bufferSize = 0;
          int lineLength = 0;
          
          while ((lineLength = getline(&buffer, &bufferSize, fp)) > 0) {
  
              struct Request* request = malloc(sizeof(struct Request));
  
              status = ares_init_options(&request->channel, &options, optmask);
              if(status != ARES_SUCCESS) {
                  printf("ares_init_options: %s\n", ares_strerror(status));
                  continue;
              }
              
              buffer[lineLength - 1] = '\0';
  
              request->domain = buffer;
              buffer = NULL;
              if (NULL == last) {
                  SLIST_INSERT_HEAD(&requests, request, next);
              } else {
                  SLIST_INSERT_AFTER(last, request, next);
              }
              last = request;
  
              ares_gethostbyname(request->channel, request->domain, AF_INET, callback, NULL);
          }
      }
  
      while (!SLIST_EMPTY(&requests)) {
          struct Request* request = SLIST_FIRST(&requests);
          wait_ares(request->channel);
          ares_destroy(request->channel);
          free(request->domain);
          SLIST_REMOVE_HEAD(&requests, next);
          free(request);
      }
  
      ares_library_cleanup();
      return 0;
  }

geoka9• 14 years ago

To be fair to Ruby, it releases the GVL when a thread is blocked on IO; so for this kind of task it wouldn't be an issue, probably.

Playing with Go: Embarrassingly Parallel Scripts

Related Stories