Re: "Connected closed by Spread" on receiving

/*
 * Applicazione per mandare richieste di transazioni alle repliche tramite SPREAD.
 * Riadattato da Lisa Vitolo da un analogo modulo in AGGRO.
 */

#include <iostream>
#include <sstream>
#include <fstream>
#include <list>
#include <map>

#include <sys/time.h>
#include <string.h>
#include <stdlib.h>
#include <errno.h>

/* Definizioni di SPREAD */
#include <sp.h>

#define GROUPNAME "default_group"

using namespace std;

/*
 * Piccola classe per gestire un messaggio inviato tramite SPREAD
 * Un messaggio e' formato da "id:numero di sequenza:nome della funzione" dove l'id è un identificatore univoco
 * della macchina e la coppia (id, numero di sequenza) identifica ogni messaggio.
 */
class Message
{
public:
    Message(int id, int seq, const string& func, const map<string, int >& accepted)
        : m_transaction(func)
    {
        if (accepted.count(m_transaction) == 1 ) {
            stringstream messageStream;
            messageStream << id << ":" << seq << ":" << m_transaction;
            m_message = messageStream.str();
        }
    }
    
    /*
     * Restituisce true se il messaggio e' stato costruito con successo.
     * Per adesso fallisce solo se la funzione non e' riconosciuta, ma in seguito potranno
     * esserci errori piu' complessi.
     */
    bool good()
    {
        return !m_message.empty();
    }
    
    const char *toString()
    {
        return m_message.c_str();
    }
    
    int byteSize()
    {
        return m_message.length() * sizeof(char);
    }
    
    string transaction()
    {
        return m_transaction;
    }
    
private:
    string m_message;
    string m_transaction;    
};

bool numberFromString(const char *, int *);
list < pair<string, int> > readTransactionsFromFile(string);
void insertAccepted(map<string, int>&);
void helpMessage(char *);

mailbox spreadConnect();
void sendMessage(int, Message& );

/*
 * Connessione al demone di SPREAD ed entrata nel gruppo.
 * Restituisce la mailbox che identifica la connessione.
 */
mailbox spreadConnect()
{
    int mbox;
    char group[1][MAX_GROUP_NAME];
    char private_group[MAX_GROUP_NAME];
    int32_t ret;

    ret = SP_connect("4803@localhost", "aggro_send", 0, 0, &mbox, private_group);

    if(ret < 0) {
        SP_error(ret);
        exit(-1);
    }

    ret = SP_join(mbox, GROUPNAME);
    
    if (ret < 0) {
        SP_error(ret);
        exit(-1);
    }
    
    return mbox;
}

/*
 * Invia un messaggio sulla mailbox data
 */
void sendMessage( int mbox, Message& mess)
{    
    int ret;
    
    ret = SP_multicast( mbox, UNRELIABLE_MESS, GROUPNAME, 1, mess.byteSize(), mess.toString());

    if(ret < 0) {
        SP_error(ret);
        exit(-1);
    }
    
    ret = SP_multicast( mbox, SAFE_MESS, GROUPNAME , 1, mess.byteSize(), mess.toString());

    if(ret < 0) {
        SP_error(ret);
        exit(-1);
    }
}

/*
 * Legge e parsa il file delle transazioni.
 */
list< pair<string, int> > readTransactionsFromFile(string filename)
{
    string line;
    ifstream stream( filename.c_str(), ifstream::in );
    list< pair<string, int> > result;
    
    if (!stream.is_open())
    {
        cerr << "Could not open \"" << filename << "\": " << strerror(errno) << endl;
        exit(-1);
    }
    
    while (!stream.eof()) {
        getline(stream, line);
        
        if (line.empty()) {
            continue;
        }
        
        /*
         * Divide nome della transazione e numero di esecuzioni
         */
        size_t pos = line.find_first_of(" ");
        if (pos == string::npos) {
            cerr << "Bad format for line \"" << line << "\". Skipped." << endl;
            continue;
        }
        
        pair<string, int > element;
        element.first = line.substr(0, pos);
        
        /*
         * Dopo lo spazio non c'e' un numero...
         */
        if (!numberFromString( line.substr(pos+1).c_str(), &(element.second) )) {
            cerr << "Bad format for line \"" << line << "\". Skipped." << endl;
            continue;
        }
        
        result.push_back(element); /* inserisce in ordine */
    }
    
    stream.close();
    return result;
}

/*
 * Notiamo che l'intero non serve a niente
 * viene usata la mappa solo per avere ricerche logaritmiche
 */
void insertAccepted(map< string, int >& m)
{
    m["read_try"] = 0;
    m["write_try"] = 0;
}

/*
 * Utility per tradurre una stringa in intero (se possibile)
 */
bool numberFromString(const char *str, int *ptr)
{
    char *error = NULL;
    int value = strtol(str, &error, 10);

    if (*error == '\0') {
        (*ptr) = value;
        return true;
    }

    return false;
}

void helpMessage( char *pname )
{
    cout << "usage: " << pname << " <machine id> [<script file>]" << endl;
    cout << "   machine id: numerical ID unique among the distributed system" << endl;
    cout << "   script file: a text file which each line in the form \"transactionname X\" where X says how much time the given"
                " transaction must be executed" << endl;
    cout << "If a script file isn't specified the transactions must be inserted one by one, with the special name \".q\" to end the list" << endl;
}

int main(int argc, char *argv[])
{
    list< pair<string, int> > transactions;
    map< string, int > acceptedTransactions;
    int seqnum = 0;
    int id = 0;
    
    
    /*
     * Nessun argomento
     */
    if (argc <= 1) {
        helpMessage(argv[0]);
        exit(0);
    }
    
    /*
     * l'ID passato come argomento non era valido
     */
    if (!numberFromString( argv[1], &id )) {
        helpMessage(argv[0]);
        exit(0);
    }
    
    mailbox mbox = spreadConnect();
    insertAccepted(acceptedTransactions);
    
    /*
     * E' stato specificato un secondo argomento, che viene trattato come il nome di un file
     * di transazioni
     */
    if (argc > 2) {
        transactions = readTransactionsFromFile( argv[2] );
        
        for (list< pair<string, int> >::iterator it = transactions.begin(); it != transactions.end(); it++) {
            for (int i = 0; i < it->second; i++) {
                string transactionName = it->first;
                
                Message mess( id, seqnum, transactionName, acceptedTransactions );
                if (!mess.good()) {
                    cerr << "Invalid transaction \"" << transactionName << "\". Skipped." << endl;
                    continue;
                }
                
                sendMessage(mbox, mess);
                seqnum++;
            }
        }
        
    } else { /* nessun argomento, si leggono le transazioni una per una da stdin */
        while (1) {
            string transactionName;
            cout << ":: Insert transaction name: ";
            cin >> transactionName;
            
            if (transactionName.compare(".q") == 0) {
                break;
            }
            
            Message mess( id, seqnum, transactionName, acceptedTransactions );
            if (!mess.good()) {
                cerr << "Invalid transaction \"" << transactionName << "\". Skipped." << endl;
                continue;
            }
            
            sendMessage(mbox, mess);
            cout << "Sent!" << endl;
            seqnum++;
        }
    }
    
    SP_disconnect(mbox);
    return 0;
}

#include <sp.h>
#include <stdlib.h>
#include <string.h>
#include <stdio.h>
#include <signal.h>
// #include "transaction.h"
// #include "orderOAB.h"

/*
 * FIXME: adatta il buffer alle dimensioni del messaggio se ci sono errori
 */
#define GROUPNAME "default_group"
#define MAX_MESS_LEN 100

char *msg = NULL;
int mbox = -1;

mailbox spreadConnect();

void sighdl(int num)
{
    free(msg);
    SP_disconnect(mbox);
    exit(-1);
}

// void shipToAggro(int, char *);

int main()
{
//     char mess1[] = "1:1:read_try";
//     char mess2[] = "2:1:write_try";
//     char mess3[] = "2:2:read_try";
//     
//     shipToAggro(0, mess1);
//     shipToAggro(0, mess2);
//     shipToAggro(0, mess3);
//     
//     char mess4[] = "1:1:read_try";
//     char mess5[] = "2:1:write_try";
//     char mess6[] = "2:2:read_try";
//     
//     shipToAggro(1, mess4);
//     shipToAggro(1, mess5);
//     shipToAggro(1, mess6);
//     
    signal(SIGINT, sighdl);
    
    mbox = spreadConnect();
    printf("Successfully connected to Spread.\n");
    
    int ret;
    int serviceType;
    short int messType;
    int endianness = 0;
    
    char sender[MAX_GROUP_NAME];
    int max_groups = 1;
    int n_groups = 0;
    char groups[1][MAX_GROUP_NAME];
    
    
    msg = (char *)malloc(sizeof(char) * MAX_MESS_LEN);
    
    while (1) {
        serviceType = messType = 0;
        memset((void *)&msg, 0, sizeof(char) * MAX_MESS_LEN);
        
        ret = SP_receive(mbox, &serviceType, sender, max_groups, &n_groups, groups, &messType, &endianness, 100 * sizeof(char), msg);
        
        if (ret < 0) {
            printf("Error in receiving: ");
            SP_error(ret);
            free(msg);
            exit(-1);
        }
        
        if (serviceType != UNRELIABLE_MESS && serviceType != SAFE_MESS) {
            continue;
        }
        
        printf("%s\n", msg);
        //shipToAggro(serviceType, msg);
    }
    
    SP_disconnect(mbox);
    free(msg);
    return 0;
}

// void shipToAggro(int type, char* mess)
// {
//     transID id;
//     transaction* t1 = (transaction *)malloc(sizeof(transaction));
//         
//     id.mID = atoi( strtok(mess, ":") );
//     id.seqnumber = atoi( strtok(NULL, ":") );
//     t1->id = id;
//     t1->funct = strtok(NULL, ":");
//     
//     if (type == 0) {
//         ord_optDelivery(t1);
//     } else {
//        // TODeliver(t1);
//     }
// }

/*
 * Connessione al demone di SPREAD ed entrata nel gruppo.
 * Restituisce la mailbox che identifica la connessione.
 */
mailbox spreadConnect()
{
    mailbox mbox;
    char group[1][MAX_GROUP_NAME];
    char private_group[MAX_GROUP_NAME];
    int32_t ret;

    ret = SP_connect("4803@localhost", "aggro_get", 0, 0, &mbox, private_group);

    if(ret < 0) {
        SP_error(ret);
        exit(-1);
    }

    ret = SP_join(mbox, GROUPNAME);
    
    if (ret < 0) {
        SP_error(ret);
        exit(-1);
    }
    
    return mbox;
}