#include <netinet/in.h>
#include <netdb.h>
#include <sys/socket.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <iostream.h>
#include <sys/types.h>
#include <sys/time.h>
#include <sys/timeb.h>
#include <unistd.h>
#include <fcntl.h>
#include <math.h>
#include <signal.h>

#include "multimandel.h"
#include "swindow.h"

// --------------------------------------------------------------------------- 

//char *slave[]={ "vulcan.solar.system", "qonos.solar.system" };
char *slave[]={
  "wman4.math.uni-wuppertal.de",
  "wmpi01.math.uni-wuppertal.de",
  "wmpi02.math.uni-wuppertal.de",
  "wmpi03.math.uni-wuppertal.de",
  "wmpi04.math.uni-wuppertal.de",
  "wmpi05.math.uni-wuppertal.de",
  "wmpi06.math.uni-wuppertal.de",
  "wmpi07.math.uni-wuppertal.de",
  "wmpi08.math.uni-wuppertal.de",
  "mathcip01.math.uni-wuppertal.de",
  "mathcip02.math.uni-wuppertal.de",
  "mathcip03.math.uni-wuppertal.de",
  "mathcip04.math.uni-wuppertal.de",
  "mathcip05.math.uni-wuppertal.de",
  "mathcip06.math.uni-wuppertal.de",
  "mathcip07.math.uni-wuppertal.de",
  "mathcip08.math.uni-wuppertal.de",
  "mathcip09.math.uni-wuppertal.de",
  "mathcip10.math.uni-wuppertal.de"
  };

static int width=800,height=600;

#if 0
static double x_1=-2.0,x_2=0.5,y_1=-0.9375,y_2=0.9375;
int maxit=1500;
static const int numcolors=32;
static const double cfac=-0.5,cphase=1.0;
static const bool pot=true;
#endif

#if 1
static double x_1=-0.7704611,x_2=-0.7661686,y_1=0.10604108,y_2=0.10926048;
static const int maxit=3500;
static const int numcolors=64;
static const double cfac=0.7,cphase=-0.5;
static const bool pot=true;
#endif

// --------------------------------------------------------------------------- 

const int num_slaves=(sizeof(slave)/sizeof(slave[0]));
int send_sock[num_slaves],master_sock,recv_sock[num_slaves];
bool alive[num_slaves];
int num_alive;
int compline[num_slaves],countlines[num_slaves];
pid_t rsh[num_slaves];

struct sockaddr_in s_in;
struct cmd_msg msgbuf;
struct data_msg *data;
size_t data_size;

double dx;
SimpleWindow *pwin;
static int (*colors)[3];

template<class T> swap(T& a, T&b) { T c=a; a=b; b*c; }

void sorry(const char *s) { extern void fini(int); perror(s); fini(0); }

void send_msgbuf(int i)
 {
   if (send(send_sock[i],&msgbuf,sizeof(msgbuf),0)<0) sorry("send");
 }

void fini(int)
 {
cout << "fini!" << endl;
   delete pwin;
   msgbuf.command='q';
   for (int i=0;i<num_slaves;++i)
      if (send_sock[i]>0) send_msgbuf(i);
   sleep(1);

   for (int i=0;i<num_slaves;++i)
    {
      if (rsh[i]>0) kill(rsh[i],SIGTERM);
      shutdown(recv_sock[i],2); close(recv_sock[i]);
      shutdown(send_sock[i],2); close(send_sock[i]);
    }
   shutdown(master_sock,2); close(master_sock);
   signal(SIGINT,SIG_DFL); raise(SIGINT);
 }

ostream & operator << (ostream &o, const struct in_addr &adr)
 {
   unsigned char c[4];
   memcpy(c,&adr,4);
   return o << (int)c[0]<<'.'<<(int)c[1]<<'.'<<(int)c[2]<<'.'<<(int)c[3];
 }

static void expose_callback(const int &w, const int &h)
  { pwin->copy_pixmap(); }

static void keypress_callback(const int &key)
 { if (key=='q') fini(0); }

static void failure_callback()
 { fini(0); }

void make_colors()
 {
   int i,j;
   double x,dx=2.0*M_PI/(numcolors-1),x2;
   colors=new int[numcolors][3];
   double const redoff=0;
   double const greenoff=cfac*M_PI/3.0;  // Normalwert: 2pi/3
   double const blueoff=2.0*greenoff;

   colors[0][0]=colors[0][1]=colors[0][2]=0;

   for (i=1,x=cphase;i<numcolors;++i,x+=dx)
    {
      x2=sin(x+redoff); if (pot) x2*=x2*x2;
      colors[i][0]=(int)(32767*(1.0+x2));
      x2=sin(x+greenoff); if (pot) x2*=x2*x2;
      colors[i][1]=(int)(32767*(1.0+x2));
      x2=sin(x+blueoff); if (pot) x2*=x2*x2;
      colors[i][2]=(int)(32767*(1.0+x2));
    }
 }

void show_one_line(int ypix, unsigned char *line)
 {
   for (int i=0;i<width;++i)
    {
      pwin->set_color(*line++);
      pwin->draw_point_pm(i,ypix);
    }
   pwin->copy_pixmap_line(ypix);
 }

int make_server_socket(int port)
 {
   int sock=socket(AF_INET,SOCK_STREAM,0);
   if (sock<0) sorry("socket1");

   memset(&s_in,0,sizeof(s_in));
   s_in.sin_family=AF_INET;
   s_in.sin_addr.s_addr=INADDR_ANY;
   s_in.sin_port=htons(MASTERPORT);
   if (bind(sock,(struct sockaddr*)&s_in,sizeof(s_in))<0) sorry("bind");
   if (listen(sock,10)<0) sorry("listen");

   return sock;
 }

int make_client_socket(const char *name, int port)
 {
   struct hostent *host=gethostbyname(name);
   if (host==0) sorry("host");

   for (int i=5;;)
    {
      int sock=socket(AF_INET,SOCK_STREAM,0);
      if (sock<0) sorry("socket");
      memset(&s_in,0,sizeof(s_in));
      s_in.sin_family=AF_INET;
      s_in.sin_addr.s_addr=((struct in_addr*)(host->h_addr))->s_addr;
      s_in.sin_port=htons(port);
      if (connect(sock,(struct sockaddr*)&s_in,sizeof(s_in))>=0) return sock;
      if (--i==0) sorry(name);
      close(sock);
      sleep(1);
    }
 }

void send_task(int to, int line)
 {
   if (to<0||to>=num_slaves) { cerr << "illegal send" << endl; exit(1); }

   msgbuf.num=to;
   msgbuf.command='c';
   msgbuf.line=line;
   msgbuf.y=y_1+line*dx;
   send_msgbuf(to);
   // cout << "sent task " << line << " to " << to << endl;
   compline[to]=line;
 }

int x_recv(int fd, void *data, int size)
 {
   char *P=(char *)data;
   int left=size,bytes;
   do
    {
      if ((bytes=recv(fd,P,left,0))<0) return 0;
      P+=bytes; left-=bytes; 
    }
   while (left>0);
   return size;
 }

void init_mandel()
 {
   if (x_2<x_1) swap(x_1,x_2);
   if (y_2<y_1) swap(y_1,y_2);
   dx=(x_2-x_1)/(double)width;
   double f=0.5*dx*height,g=0.5*(y_1+y_2);
   y_1=g-f; y_2=g+f;
 }

int main(int argc, char **argv)
 {
   int i;

   struct timeb t1,t2,t3;
   ftime(&t1);
   signal(SIGINT,fini);
   init_mandel();

   if (argc>=2 && argv[1][0]=='+')
      for (i=0;i<num_slaves;++i)
       {
         if ((rsh[i]=fork())==0)
          {
            char buffer[80];
            execlp("rsh","rsh","-n",slave[i],"/usr/tmp/slave",0);
          }
       }

   msgbuf.x1=x_1; msgbuf.x2=x_2;
   msgbuf.width=width; msgbuf.depth=maxit;
   msgbuf.line=numcolors;
   data_size=sizeof(data_msg)+sizeof(unsigned char[width]);
   data=(data_msg*)new char[data_size];

   master_sock=make_server_socket(MASTERPORT);

   for (i=0;i<num_slaves;++i)
    {
      send_sock[i]=make_client_socket(slave[i],SLAVEPORT);
      msgbuf.num=i;
      msgbuf.command='s';
      send_msgbuf(i);
      // cout << "sent to " << i << endl;
    }

   num_alive=0;
   for (i=0;i<num_slaves;++i)
    {
      memset(&s_in,0,sizeof(s_in));
      int addrlen=sizeof(struct sockaddr_in);
      int sock=accept(master_sock,(struct sockaddr*)&s_in,&addrlen);
      if (sock<0) sorry("accept");

      cout << "connected to " << s_in.sin_addr << endl;

      if (x_recv(sock,data,data_size)==0) sorry("recv");
      // cout << "received hello from " << data->num 
      //      << " (value " << data->it[0] << ")\n";

      recv_sock[data->num]=sock;
      if (alive[data->num]) exit(99);
      alive[data->num]=true;
      ++num_alive;

      send_task(data->num,i);
    }

   if (num_alive==0) { cerr << "nobody wants to talk to me" << endl; exit(1); }

   make_colors();
   pwin=new SimpleWindow(width,height,"MandelMaster",numcolors,colors);
   pwin->set_expose_func(expose_callback);
   pwin->set_keypress_func(keypress_callback);

   int lines_to_receive=height, next_line=num_slaves-1;

   ftime(&t2);

   do
    {
      fd_set wait_set;
      FD_ZERO(&wait_set);
      int max=0;
      for (i=0;i<num_slaves;++i) 
         if (alive[i])
          {
            FD_SET(recv_sock[i],&wait_set);
            if (recv_sock[i]>max) max=recv_sock[i];
          }

      struct timeval tv_wait;
      tv_wait.tv_sec=20; tv_wait.tv_usec=0;
      if (select(max+1,&wait_set,0,0,&tv_wait)==0)
         cerr << "warning: no message in 20 seconds" << endl;

      for (i=0;i<num_slaves;++i)
         if (FD_ISSET(recv_sock[i],&wait_set))
          {
            int bytes=x_recv(recv_sock[i],data,data_size);
            if (bytes<=0)
             {
               cerr << i << " died" << endl;
               alive[i]=false;
               if (--num_alive==0) { cerr << "all dead" << endl; exit(1); }
             }
            else
             {
               // cout << "message from " << data->num << endl;
               if (data->num>=num_slaves)
                  cerr << "something's obviously wrong..." << endl;
               else 
                {
                  if ( bytes!=data_size || data->line!=compline[data->num] )
                   {
                     cerr << "received garbage (" << bytes << "<->"
                          << data_size << ") "
                          << data->line << "<->" << compline[data->num]
                          << endl;
                     send_task(data->num,compline[data->num]);
                   }
                  else
                   {
                     // cout << "received line " << data->line << endl;
                     ++countlines[data->num];
                     if (--lines_to_receive>0 && ++next_line<height)
                        send_task(data->num,next_line);
                     show_one_line(data->line,data->it);
                     if (lines_to_receive==0) break;
                   }
                }
             }
          }

      pwin->handle_events(false);
    }
   while (lines_to_receive>0);

   ftime(&t3);
   cout << "\nsuccessfully completed!\n";
   for (i=0;i<num_slaves;++i)
      cout << "received " << countlines[i] << " lines from "
           << slave[i] << endl;
   cout << "time: " << (t3.time-t1.time)+(t3.millitm-t1.millitm)/1000.0 << " s ("
        << (t3.time-t2.time)+(t3.millitm-t2.millitm)/1000.0 << " s computing time)" << endl;

   pwin->handle_events(true);
   fini(0);
 }