TwitterEdge.h

Go to the documentation of this file.

#ifndef _TWITTER_EDGE_
#define _TWITTER_EDGE_
 
#include <iostream>
#include <ctime>
#include "CombBLAS/CombBLAS.h"
 
struct DetSymmetricize;
 
class TwitterEdge
{
public:
    TwitterEdge(): count(0), follower(0), latest(0) {};
    template <typename X>
    TwitterEdge(X x):count(0), follower(0), latest(0) {};   // any upcasting constructs the default object too
 
    TwitterEdge(short mycount, bool myfollow, time_t mylatest):count(mycount), follower(myfollow), latest(mylatest) {};
    bool isFollower() const {   return follower; };
    bool isRetwitter() const {  return (count > 0); };
    bool TweetWithinInterval (time_t begin, time_t end) const   {   return ((count > 0) && (begin <= latest && latest <= end));  };
    bool TweetSince (time_t begin) const    {   return ((count > 0) && (begin <= latest));  };
    bool LastTweetBy (time_t end) const {   return ((count > 0) && (latest <= end));  };
 
    operator bool () const  {   return true;    } ;       // Type conversion operator (ABAB: Shoots in the foot by implicitly converting many things)
 
    TwitterEdge & operator+=(const TwitterEdge & rhs) 
    {
        std::cout << "Error: TwitterEdge::operator+=() shouldn't be executed" << std::endl;
        count += rhs.count;
        follower |= rhs.follower;
        if(rhs.count > 0)   // ensure that addition with additive identity doesn't change "latest"
            latest = std::max(latest, rhs.latest);
        return *this;
    }
    bool operator ==(const TwitterEdge & b) const
    {
        return ((follower == b.follower) && (latest == b.latest) &&  (count == b.count));
    }
 
    friend std::ostream& operator<<( std::ostream& os, const TwitterEdge & twe);
    friend TwitterEdge operator*( const TwitterEdge & a, const TwitterEdge & b);
 
private:
    bool follower;      // default constructor sets all to zero
    time_t latest;      // not assigned if no retweets happened
    short count;        
    
    template <typename IT>
    friend class TwitterReadSaveHandler;
    
    friend struct DetSymmetricize;
};
 
std::ostream& operator<<(std::ostream& os, const TwitterEdge & twe )    
{      
    if( twe.follower == 0 && twe.latest == 0 &&  twe.count == 0)
        os << 0;
    else
        os << 1;
    return os;    
};
 
TwitterEdge operator*( const TwitterEdge & a, const TwitterEdge & b)
{
    // One of the parameters is an upcast from bool (used in Indexing), so return the other one
    if(a == TwitterEdge())  return b;
    else    return a;
}
 
 
template <class IT>
class TwitterReadSaveHandler
{
    public:
        TwitterReadSaveHandler() {};
        TwitterEdge getNoNum(IT row, IT col) { return TwitterEdge(); }
 
        MPI_Datatype getMPIType()
        {
            return combblas::MPIType<TwitterEdge>(); // utilize the MPI type cache
        }
 
        void binaryfill(FILE * rFile, IT & row, IT & col, TwitterEdge & val)
        {
            TwitterInteraction twi;
            size_t entryLength = fread (&twi,sizeof(TwitterInteraction),1,rFile);
            row = twi.from - 1 ;
            col = twi.to - 1;
            val = TwitterEdge(twi.retweets, twi.follow, twi.twtime); 
            if(entryLength != 1)
                std::cout << "Not enough bytes read in binaryfill " << std::endl;
        }
        size_t entrylength() { return sizeof(TwitterInteraction); }
 
        template <typename c, typename t>
        TwitterEdge read(std::basic_istream<c,t>& is, IT row, IT col)
        {
            TwitterEdge tw;
            is >> tw.follower;
            is >> tw.count;
            if(tw.count > 0)
            {
                std::string date;
                std::string time;
                is >> date;
                is >> time;
 
                struct tm timeinfo;
                int year, month, day, hour, min, sec;
                sscanf (date.c_str(),"%d-%d-%d",&year, &month, &day);
                sscanf (time.c_str(),"%d:%d:%d",&hour, &min, &sec);
 
                memset(&timeinfo, 0, sizeof(struct tm));
                timeinfo.tm_year = year - 1900; // year is "years since 1900"
                timeinfo.tm_mon = month - 1 ;   // month is in range 0...11
                timeinfo.tm_mday = day;         // range 1...31
                timeinfo.tm_hour = hour;        // range 0...23
                timeinfo.tm_min = min;          // range 0...59
                timeinfo.tm_sec = sec;          // range 0.
                tw.latest = timegm(&timeinfo);
                if(tw.latest == -1) { std::cout << "Can not parse time date" << std::endl; exit(-1);}
            }
            else
            {
                tw.latest = 0;  // initialized to dummy
            }
            //cout << row << " follows " << col << "? : " << tw.follower << " and the retweet count is " << tw.count << endl;
            return tw;
        }
        
    
        template <typename c, typename t>
        void save(std::basic_ostream<c,t>& os, const TwitterEdge & tw, IT row, IT col)  // save is NOT compatible with read
        {
            os << row << "\t" << col << "\t";
            os << tw.follower << "\t";
            os << tw.count << "\t";
            os << tw.latest << std::endl;
        }
    private:
        struct TwitterInteraction
        {
                int32_t from;
                int32_t to;
                bool follow;
                int16_t retweets;
                time_t twtime;
        };
};
 
 
struct ParentType
{
    ParentType():id(-1) { };
    ParentType(int64_t myid):id(myid) { };
    int64_t id;
    bool operator ==(const ParentType & rhs) const
    {
        return (id == rhs.id);
    }
    bool operator !=(const ParentType & rhs) const
    {
        return (id != rhs.id);
    }
    ParentType & operator+=(const ParentType & rhs) 
    {
        std::cout << "Adding parent with id: " << rhs.id << " to this one with id " << id << std::endl;
        return *this;
    }
    const ParentType operator++(int)    // for iota
    {
        ParentType temp(*this); // post-fix requirement
        ++id;
        return temp;
    }
    friend std::ostream& operator<<(std::ostream& os, const ParentType & twe ); 
 
    template <typename IT>
    friend ParentType operator+( const IT & left, const ParentType & right); 
};
 
std::ostream& operator<<(std::ostream& os, const ParentType & twe )    
{      
    os << "Parent=" << twe.id;
    return os;    
};
 
ParentType NumSetter(ParentType & num, int64_t index) 
{
    return ParentType(index);
}
 
template <typename IT>
ParentType operator+( const IT & left, const ParentType & right)
{
    return ParentType(left+right.id);
}
 
// forward declaration
template <typename SR, typename T>
void select2nd(void * invec, void * inoutvec, int * len, MPI_Datatype *datatype);
 
 
template <typename SR, typename VECTYPE>
static VECTYPE filtered_select2nd(const TwitterEdge & arg1, const VECTYPE & arg2, time_t & sincedate)  
{
    if(sincedate == -1) // uninitialized
    {
        struct tm timeinfo;
        memset(&timeinfo, 0, sizeof(struct tm));
        int year, month, day, hour, min, sec;
        year = 2009;    month = 7;  day = 1;
        hour = 0;   min = 0;    sec = 0;
        
        timeinfo.tm_year = year - 1900; // year is "years since 1900"
        timeinfo.tm_mon = month - 1 ;   // month is in range 0...11
        timeinfo.tm_mday = day;         // range 1...31
        timeinfo.tm_hour = hour;        // range 0...23
        timeinfo.tm_min = min;          // range 0...59
        timeinfo.tm_sec = sec;          // range 0.
        sincedate = timegm(&timeinfo);
        
        std::ostringstream outs;
        outs << "Initializing since date (only once) to " << sincedate << std::endl;
        combblas::SpParHelper::Print(outs.str());
    }
    
    if(arg1.isRetwitter() && arg1.LastTweetBy(sincedate))   // T1 is of type edges for BFS
    {
        return arg2;
    }
    else
    {
        SR::returnedSAID(true);
        return VECTYPE();   
        // return null-type parent id (for BFS) or 
        // double() for MIS - POD objects are zero initilied
    }
}
 
 
 
struct LatestRetwitterBFS
{
    static MPI_Op MPI_BFSADD;
    static ParentType id() { return ParentType(); } // additive identity
 
    // the default argument means that this function can be used like this:
    // if (returnedSAID()) {...}
    // which is how it is called inside CombBLAS routines. That call conveniently clears the flag for us.
    static bool returnedSAID(bool setFlagTo = false) 
    {
        static bool flag = false;
 
        bool temp = flag; // save the current flag value to be returned later. Saves an if statement.
        flag = setFlagTo; // set/clear the flag.
        return temp;
    }
 
    static ParentType add(const ParentType & arg1, const ParentType & arg2)
    {
        return ((arg2 == ParentType()) ? arg1: arg2);
    }
 
    static MPI_Op mpi_op() 
    { 
        MPI_Op_create(select2nd<LatestRetwitterBFS,ParentType>, false, &MPI_BFSADD);    // \todo {do this once only, by greating a MPI_Op buffer}
        return MPI_BFSADD;
    }
    static time_t sincedate;
    static ParentType multiply(const TwitterEdge & arg1, const ParentType & arg2)
    {
        return filtered_select2nd<LatestRetwitterBFS>(arg1, arg2, sincedate);
    }
    static void axpy(TwitterEdge a, const ParentType & x, ParentType & y)
    {
        y = add(y, multiply(a, x));
    }
};
 
time_t LatestRetwitterBFS::sincedate = -1;
 
// select2nd for doubles
template <typename SR, typename T>
void select2nd(void * invec, void * inoutvec, int * len, MPI_Datatype *datatype)
{
    T * pinvec = static_cast<T*>(invec);
    T * pinoutvec = static_cast<T*>(inoutvec);
    for (int i = 0; i < *len; i++)
    {
        pinoutvec[i] = SR::add(pinvec[i], pinoutvec[i]);
    }
}
 
 
MPI_Op LatestRetwitterBFS::MPI_BFSADD;
 
struct getfringe: public std::binary_function<ParentType, ParentType, ParentType>
{
    ParentType operator()(ParentType x, const ParentType & y) const
    {
        return x;
    }
    
};
 
// x: Parent type (always 1 if exits, sparse)
// y: degree (dense)
struct seldegree: public std::binary_function<ParentType, int64_t, int64_t>
{
    int64_t operator()(ParentType x, const int64_t & y) const
    {
        return y;
    }
    
};
 
// This is like an "isparentset" with the extra parameter that we don't care
struct passifthere: public std::binary_function<ParentType, int64_t, bool>
{
    bool operator()(ParentType x, const int64_t & y) const
    {
        return  (x != ParentType());
    }
    
};
 
// DoOp for MIS's EWiseApply
struct is2ndSmaller: public std::binary_function<double, double, bool>
{
    bool operator()(double m, double c) const
    {
        return (c < m);
    }
};
 
// BinOp for MIS's EWiseApply
struct return1_uint8: public std::binary_function<double, double, uint8_t>
{
    uint8_t operator() (double t1, double t2)
    {
        return (uint8_t) 1;
    }
};
 
 
// x: elements from fringe (sparse), y: elements from parents (dense) 
// return true for edges that are not filtered out, and not previously discovered
// if the edge was filtered out, then x would be ParentType()
// if y was already discovered its parent would NOT be ParentType()
struct keepinfrontier_f: public std::binary_function<ParentType, ParentType, bool>
{
    bool operator()(ParentType x, const ParentType & y) const
    {
        return ( x != ParentType() && y == ParentType()) ;
    }
    
};
 
struct isparentset: public std::unary_function<ParentType, bool>
{
    bool operator()(const ParentType & x) const
    {
        return ( x != ParentType() ) ;
    }
    
};
 
// Matrix type: TwitterEdge
// Vector type: double
struct LatestRetwitterMIS
{
    static double id() { return 0.0; }  // additive identity
    
    // the default argument means that this function can be used like this:
    // if (returnedSAID()) {...}
    // which is how it is called inside CombBLAS routines. That call conveniently clears the flag for us.
    static bool returnedSAID(bool setFlagTo = false) 
    {
        static bool flag = false;
        
        bool temp = flag; // save the current flag value to be returned later. Saves an if statement.
        flag = setFlagTo; // set/clear the flag.
        return temp;
    }
    
    static double add(const double & arg1, const double & arg2)
    {
        return std::min(arg1, arg2);
    }
    
    static MPI_Op mpi_op() 
    { 
        return MPI_MIN;
    }
    static time_t sincedate;
    static double multiply(const TwitterEdge & arg1, const double & arg2)  // filtered select2nd
    {
        return filtered_select2nd<LatestRetwitterMIS>(arg1, arg2, sincedate);
    }
    static void axpy(TwitterEdge a, const double & x, double & y)
    {
        y = add(y, multiply(a, x));
    }
};
 
// Matrix type: TwitterEdge
// Vector type: double
struct LatestRetwitterSelect2nd // also used for finding neighbors of the candidate set in MIS
{
    static MPI_Op MPI_SEL2NDADD;
    static double id() { return 0.0; }  // additive identity
    
    // the default argument means that this function can be used like this:
    // if (returnedSAID()) {...}
    // which is how it is called inside CombBLAS routines. That call conveniently clears the flag for us.
    static bool returnedSAID(bool setFlagTo = false) 
    {
        static bool flag = false;
        
        bool temp = flag; // save the current flag value to be returned later. Saves an if statement.
        flag = setFlagTo; // set/clear the flag.
        return temp;
    }
    
    static double add(const double & arg1, const double & arg2)
    {
        return arg2;
    }
    
    static MPI_Op mpi_op() 
    { 
        MPI_Op_create(select2nd<LatestRetwitterSelect2nd,double>, false, &MPI_SEL2NDADD);   // \todo {do this once only, by greating a MPI_Op buffer}
        return MPI_SEL2NDADD;
    }
    static time_t sincedate;
    static double multiply(const TwitterEdge & arg1, const double & arg2)  // filtered select2nd
    {
        return filtered_select2nd<LatestRetwitterSelect2nd>(arg1, arg2, sincedate);
    }
    static void axpy(TwitterEdge a, const double & x, double & y)
    {
        y = add(y, multiply(a, x));
    }
};
 
time_t LatestRetwitterMIS::sincedate = -1;
time_t LatestRetwitterSelect2nd::sincedate = -1;
MPI_Op LatestRetwitterSelect2nd::MPI_SEL2NDADD;
 
 
 
#endif