// $Id$ -*- c++ -*-

#include <iostream>
#include "pvm3.h"

#include "pipeline.h"

#include <unistd.h>

// Calculates ln x using its Taylor polynomial with center point 1
// see http://en.wikipedia.org/wiki/Taylor_series#List_of_Maclaurin_series_of_some_common_functions

void calculate (float x, float &y, float &s, int k)
{
    // Delay to make measuring easier
    for (int i = 0; i < 1000 * 1000 * 10; ++i)
        ;
    
    if (k == 1)
    {
        y = x - 1;
        s = y;
    } else {    
        y *= -(x - 1);
        s += y / (float)k;
    }
}

int main (int argc, char** argv)
{
    // Get neighbour addresses
    int tidParent = pvm_parent ();
    if (tidParent == PvmNoParent)
    {
        std::cerr << "worker started standalone" << std::endl;
        return pvm_exit ();
    }

    pvm_recv (tidParent, MSG_INIT);
    int iSelf;
    int tidIn;
    int tidOut;
    pvm_upkint (&iSelf, 1, 1);
    pvm_upkint (&tidIn, 1, 1);
    pvm_upkint (&tidOut, 1, 1);

    // The actual implementation
    bool fDone = false;
    while (!fDone)
    {
        int bufInput = pvm_recv (tidIn, -1);

        int cByte, tag, tidSender;
        pvm_bufinfo (bufInput, &cByte, &tag, &tidSender);

        switch (tag)
        {
        case MSG_DATA:        
            float x;
            float y;
            float s;
            
            pvm_upkfloat (&x, 1, 1);
            if (tidIn != tidParent)
            {
                pvm_upkfloat (&y, 1, 1);
                pvm_upkfloat (&s, 1, 1);
            }

            std::cout << iSelf << "\t" << y << std::endl;
            
            calculate (x, y, s, iSelf + 1);
            
            pvm_initsend (PvmDataDefault);
            if (tidOut != tidParent)
            {
                pvm_pkfloat (&x, 1, 1);
                pvm_pkfloat (&y, 1, 1);
            }
            pvm_pkfloat (&s, 1, 1);
            pvm_send (tidOut, MSG_DATA);
            break;

        case MSG_DONE:
            fDone = true;
            pvm_initsend (PvmDataDefault);
            pvm_send (tidOut, MSG_DONE);
            break;
        }
    }
        
    return pvm_exit();
}