parar.h

Go to the documentation of this file.

/*
  This file is part of MADNESS.

  Copyright (C) 2007,2010 Oak Ridge National Laboratory

  This program is free software; you can redistribute it and/or modify
  it under the terms of the GNU General Public License as published by
  the Free Software Foundation; either version 2 of the License, or
  (at your option) any later version.

  This program is distributed in the hope that it will be useful,
  but WITHOUT ANY WARRANTY; without even the implied warranty of
  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
  GNU General Public License for more details.

  You should have received a copy of the GNU General Public License
  along with this program; if not, write to the Free Software
  Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA

  For more information please contact:

  Robert J. Harrison
  Oak Ridge National Laboratory
  One Bethel Valley Road
  P.O. Box 2008, MS-6367

  email: harrisonrj@ornl.gov
  tel:   865-241-3937
  fax:   865-572-0680

  $Id$
*/
#ifndef MADNESS_WORLD_PARAR_H__INCLUDED
#define MADNESS_WORLD_PARAR_H__INCLUDED


#include <madness/world/archive.h>
#include <madness/world/binfsar.h>
#include <madness/world/worldfwd.h>
#include <madness/world/worldgop.h>

#include <unistd.h>
#include <cstring>
#include <cstdio>

namespace madness {
    namespace archive {

        class ParallelSerializableObject {};


        template <typename Archive>
        class BaseParallelArchive {
            World* world;       
            mutable Archive ar; 
            int nio;            
            bool do_fence;      
            char fname[256];    
            int nclient;        

        public:
            static const bool is_parallel_archive = true;

            BaseParallelArchive() : world(0), ar(), nio(0), do_fence(true) {}


            ProcessID io_node(ProcessID rank) const {
                return rank%nio;
            }

            ProcessID my_io_node() const {
                MADNESS_ASSERT(world);
                return io_node(world->rank());
            }

            int num_io_clients() const {
                MADNESS_ASSERT(world);
                return nclient;
            }

            bool is_io_node() const {
                MADNESS_ASSERT(world);
                return world->rank() == my_io_node();
            }

            World* get_world() const {
                MADNESS_ASSERT(world);
                return world;
            }


            void open(World& world, const char* filename, int nwriter=1) {
                this->world = &world;
                nio = nwriter;
#if defined(HAVE_IBMBGP) || defined(HAVE_IBMBGQ)
                /* Jeff believes that BG is designed to handle up to *
                 * one file per node and I assume no more than 8 ppn */
                int maxio = world.size()/8;
#else
                int maxio = 50;
#endif
                if (nio > maxio) nio = maxio; // Sanity?
                if (nio > world.size()) nio = world.size();

                MADNESS_ASSERT(filename);
                MADNESS_ASSERT(strlen(filename)-1<sizeof(fname));
                strcpy(fname,filename); // Save the filename for later
                char buf[256];
                MADNESS_ASSERT(strlen(filename)+7 <= sizeof(buf));
                sprintf(buf, "%s.%5.5d", filename, world.rank());

                if (world.rank() == 0) {
                    ar.open(buf);
                    ar & nio; // read/write nio from/to the archive
                    MADNESS_ASSERT(nio <= world.size());
                }

                // Ensure all agree on value of nio that may also have changed if reading
                world.gop.broadcast(nio, 0);

                // Other reader/writers can now open the local archive
                if (is_io_node() && world.rank()) {
                    ar.open(buf);
                }

                // Count #client
                ProcessID me = world.rank();
                nclient=0;
                for (ProcessID p=0; p<world.size(); ++p) if (io_node(p) == me) ++nclient;

//                 if (is_io_node()) {
//                     madness::print("I am an IO node with",nclient,"clients and file",buf);
//                 }
//                 else {
//                     madness::print("I am a client served by",my_io_node(),fname);
//                 }
            }

            static bool exists(World& world, const char* filename) {
                char buf[256];
                MADNESS_ASSERT(strlen(filename)+7 <= sizeof(buf));
                sprintf(buf, "%s.%5.5d", filename, world.rank());
                bool status;
                if (world.rank() == 0)
                    status = (access(buf, F_OK|R_OK) == 0);

                world.gop.broadcast(status);

                return status;
            }

            void close() {
                MADNESS_ASSERT(world);
                if (is_io_node()) ar.close();
            }

            Archive& local_archive() const {
                MADNESS_ASSERT(world);
                MADNESS_ASSERT(is_io_node());
                return ar;
            }

            template <typename objT>
            void broadcast(objT& obj, ProcessID root) const {
                get_world()->gop.broadcast_serializable(obj, root);
            }


            static void remove(World& world, const char* filename) {
                if (world.rank() == 0) {
                    char buf[256];
                    MADNESS_ASSERT(strlen(filename)+7 <= sizeof(buf));
                    for (ProcessID p=0; p<world.size(); ++p) {
                        sprintf(buf, "%s.%5.5d", filename, p);
                        if (::remove(buf)) break;
                    }
                }
            }

            void remove() {
                MADNESS_ASSERT(world);
                remove(*world, fname);
            }


            bool dofence() const {
                return this->do_fence;
            }

            void set_dofence(bool dofence) {
                do_fence = dofence;
            }
        };



        class ParallelOutputArchive : public BaseParallelArchive<BinaryFstreamOutputArchive>, public BaseOutputArchive {
        public:
            ParallelOutputArchive() {}

            ParallelOutputArchive(World& world, const char* filename, int nio=1)  {
                open(world, filename, nio);
            }

            void flush() {
                if (is_io_node()) local_archive().flush();
            }
        };


        class ParallelInputArchive : public BaseParallelArchive<BinaryFstreamInputArchive>, public  BaseInputArchive {
        public:
            ParallelInputArchive() {}

            ParallelInputArchive(World& world, const char* filename, int nio=1) {
                open(world, filename, nio);
            }
        };


        template <class T>
        struct ArchivePrePostImpl<ParallelOutputArchive,T> {
            static void preamble_store(const ParallelOutputArchive& ar) {};
            static inline void postamble_store(const ParallelOutputArchive& ar) {};
        };

        template <class T>
        struct ArchivePrePostImpl<ParallelInputArchive,T> {
            static inline void preamble_load(const ParallelInputArchive& ar) {};
            static inline void postamble_load(const ParallelInputArchive& ar) {};
        };


        template <class T>
        struct ArchiveImpl<ParallelOutputArchive, T> {
            template <typename Q>
            static inline
            typename madness::enable_if<is_derived_from<Q, ParallelSerializableObject>, const ParallelOutputArchive&>::type
            wrap_store(const ParallelOutputArchive& ar, const Q& t) {
                ArchiveStoreImpl<ParallelOutputArchive,T>::store(ar,t);
                return ar;
            }

            template <typename Q>
            static inline
            typename madness::disable_if<is_derived_from<Q, ParallelSerializableObject>, const ParallelOutputArchive&>::type
            wrap_store(const ParallelOutputArchive& ar, const Q& t) {
                if (ar.get_world()->rank()==0) {
                    ar.local_archive() & t;
                }
                return ar;
            }
        };

        template <class T>
        struct ArchiveImpl<ParallelInputArchive, T> {
            template <typename Q>
            static inline
            typename madness::enable_if<is_derived_from<Q, ParallelSerializableObject>, const ParallelInputArchive&>::type
            wrap_load(const ParallelInputArchive& ar, const Q& t) {
                ArchiveLoadImpl<ParallelInputArchive,T>::load(ar,const_cast<T&>(t));
                return ar;
            }

            template <typename Q>
            static inline
            typename madness::disable_if<is_derived_from<Q, ParallelSerializableObject>, const ParallelInputArchive&>::type
            wrap_load(const ParallelInputArchive& ar, const Q& t) {
                if (ar.get_world()->rank()==0) {
                    ar.local_archive() & t;
                }
                ar.broadcast(const_cast<T&>(t), 0);
                return ar;
            }
        };


        template <class T>
        struct ArchiveImpl< ParallelOutputArchive, archive_array<T> > {
            static inline const ParallelOutputArchive& wrap_store(const ParallelOutputArchive& ar, const archive_array<T>& t) {
                if (ar.get_world()->rank() == 0) ar.local_archive() & t;
                return ar;
            }
        };

        template <class T>
        struct ArchiveImpl< ParallelInputArchive, archive_array<T> > {
            static inline const ParallelInputArchive& wrap_load(const ParallelInputArchive& ar, const archive_array<T>& t) {
                if (ar.get_world()->rank() == 0) ar.local_archive() & t;
                ar.broadcast(t, 0);
                return ar;
            }
        };

        template <class T, std::size_t n>
        struct ArchiveImpl<ParallelOutputArchive, T[n]> {
            static inline const ParallelOutputArchive& wrap_store(const ParallelOutputArchive& ar, const T(&t)[n]) {
                ar << wrap(&t[0],n);
                return ar;
            }
        };

        template <class T, std::size_t n>
        struct ArchiveImpl<ParallelInputArchive, T[n]> {
            static inline const ParallelInputArchive& wrap_load(const ParallelInputArchive& ar, const T(&t)[n]) {
                ar >> wrap(&t[0],n);
                return ar;
            }
        };
    }
}

#endif // MADNESS_WORLD_PARAR_H__INCLUDED