CUDA (Compute Unified Device Architecture) is a parallel computing architecture devel- oped by NVIDIA. Simply put, CUDA is the compute engine in NVIDIA graphics processing units or GPUs, that is accessible to software developers through industry standard program- ming languages. Programmers use C for CUDA, compiled through a PathScale Open64 C compiler, to code algorithms for execution on the GPU. CUDA is architected to support all computational interfaces, including C and new open standards like OpenGL and DX11 Compute.
Our MMR cluster also includes a CUDA library, so it also supports a CUDA cluster with the combination of MPI and CUDA. The easiest way to combine MPI and CUDA is to use nvcc to compile MPI-enabled CUDA codes with a nvcc compiler wrapper. The nvcc compiler wrapper is somewhat more complex than the typical mpicc compiler wrapper, however, for the users it is easier to make MPI code into CUDA source files and compile with nvcc than the other way around. Blow is what a makefile for a MPI-enabled CUDA code would look like: MPICC := nvcc -Xptxas -v MPI_INCLUDES := /usr/include/mpi MPI_LIBS := /usr/lib %.o : %.cu $(MPICC) -I$(MPI_INCLUDES) -o $@ -c $< program_name : program_name.o $(MPICC) -L$(MPI_LIBS) -o $@ *.o
clean :
rm program_name.o
6
Conclusions
Scalable computational developers are in critical need of scalable development platforms that can automatically take advantage of distributed compute resources. The MapReduce model has recently gained popularity and the backing of major industry players as a viable option to quickly and efficiently build scalable distributed applications and has been adopted as a core technology by the top two web search companies-Google and Yahoo!.
In this thesis, we introduced a new open-source implementation–MPI MapReduce (MMR )–which significantly outperforms the leading open-source solution Hadoop. We showed that, unlike Hadoop, we can efficiently and predictably scale up a MapReduce computation. Specifically, for CPU-bound processing, our platform provides linear scaling with respect to the number and speed of CPUs provided. For the wordcount application, which is closely related to common indexing tasks, we demonstrated super-linear speedup, whereas for I/O- bound and memory-constrained tasks the speedup is substantial but sub-linear.
Based on this initial evaluation, and our development experience, we can confidently conclude that the proposed MMR platform provides a promising basis for the development of large-scale computational processing tools. Our next short term goals are to scale up the experiments to tens/hundreds of cores, and develop specific tools that can deal with actual terabyte-size targets in real-time using a GPU MMR MapReduce model.
In addition to the MMR programming model, we have also designed a MMR live DVD which facilitates the automatic installation and configuration of a Linux cluster with inte-
grated MMR libraries. The cluster is simple but the software environment that it provides is sufficient for developing and running MMR applications, and there is no need to manu- ally install the software on each node. Moreover, users can easily build such a cluster with various types of machines, from a desktop PC to a Mac laptop, with Intel or compatible processors. This provides a convenient and economic MMR cluster solution to companies and institutions who would like to build such a cluster with no extra hardware costs.
Bibliography
[1] http://en.wikipedia.org/wiki/Moore%27s_law. [2] http://hadoop.apache.org/core/.
[3] M. Mitzenmatcher A. Broder. Network applications of bloom filters: a survey. In In Annual Allerton Conference on Communication, Control, and Computing, Urbana- Champaign, Illinois, USA, 2002.
[4] P. van Oorschot A. Menezes and S. Vanstone. Handbook of Applied Cryptography. CRC Press, Inc, 1997.
[5] Maarten van Steen Andrew S. Tanenbaum. Distributed Systems Principles and
Paradigms. Pearson Education, 2002.
[6] R. Rubinfeld A. Tal. B. Chazelle, J. Kilian. The bloomier filter an efficient data structure for static support lookup tables. In In Proceedings of the 15th Annual ACM-SIAM
Symposium on Discrete Algorithms, pages 30–39, 2004.
[7] Intel Corporation. Preboot Execution Environment Specification. SystemSoft, 1999. [8] Ghemawat S. Dean, J. Mapreduce: Simplified data processing on large clusters. In n
Proceedings of the Sixth Symposium on Operating System Design and Implementation,
San Francisco, CA, 2004.
[9] Legnago (VR) G. Argentini, R. Group. A generalization of amdahl’s law and relative conditions of parallelism. In Distributed, Parallel, and Cluster Computing, 2002.
[10] Gobioff H. Ghemawat, S. and S. Leung. The google file system. In In 19th ACM Sym-
posium on Operating Systems Principles, Proceedings, pages 2943. ACM Press, 2003.
[11] Joseph A. Goguen and Jose Meseguer. Unifying functional, object-oriented and rela-
tional programming with logical semantics. MIT Press, 1987.
[12] Roger K. Blashfield Mark S. Aldenderfer. Cluster Analysis. Sage, 1984.
[13] D. Patterson. Latency lags bandwidth. In I Communications of the ACM, volume 47, 2004.
[14] Raghuraman R. Penmetsa A. Bradski G. Ranger, C. and C. Kozyrakis. Evaluating mapreduce for multi-core and multiprocessor systems. In Proceedings of the 13th Intl.
Symposium on High-Performance Computer Architecture, Phoenix, AZ,, 2007.
[15] William Stallings. Operating Systems Internals and Design Principles. Pearson Prentice Hall, 2004.
[16] Chris Tofallis. Investment volatility: A critique of standard beta estimation and a simple way forward. In European Journal of Operational Research, volume 187, pages 1358–1367, 2008.
[17] Ewing Lusk William Gropp and Thomas Sterling. Beowulf Cluster Computing with
Appendix A C Programming Example // I n c l u d e n e c e s s a r y h e a d e r f i l e s h e r e . /* * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * */ # i n c l u d e " mmr . h " // A h e a d e r f i l e for the MMR f u n c t i o n s . # i n c l u d e " c o m m o n _ m a c r o s . h " // A h e a d e r f i l e for s o m e c o m m o n m a c r o s . /* * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * */ // F o l l o w i n g f u n c t i o n s m u s t be i m p l e m e n t e d by the u s e r . /* * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * */ // Key c o m p a r i s o n f u n c t i o n . int k e y C o m p ( c o n s t v o i d * key1 , c o n s t v o i d * k e y 2 ) { // R e t u r n 0 o n l y w h e n two k e y s are i d e n t i c a l . /* F o l l o w i n g is a w o r d c o u n t e x a m p l e w h i c h c o m p a r e s two k e y s as two w o r d s . r e t u r n s t r c m p (( c o n s t c h a r *) key1 , ( c o n s t c h a r *) k e y 2 ) ; */ } /* * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * */ // Map f u n c t i o n . v o i d map ( m a p _ a r g s _ t * a r g s ) { a s s e r t ( a r g s ) ;
a s s e r t ( args - > length >0) ;
c h a r * d a t a = ( c h a r *) args - > d a t a ; a s s e r t ( d a t a ) ;
// F i n d or c a l c u l a t e k e y s and v a l u e s f r o m d a t a and e m i t a key / v a l u e p a i r for e a c h key .
// C a l l e m i t _ i n t e r m e d i a t e () ( see M a p R e d u c e S c h e d u l e r . h ) . It can be c a l l e d m u l t i p l e t i m e s d e p e n d i n g on the n u m b e r of key / v a l u e p a i r s .
/* F o l l o w i n g is a w o r d c o u n t e x a m p l e w h i c h e m i t s a w o r d as a key and 1 i . e . the c o u n t as a v a l u e .
e m i t _ i n t e r m e d i a t e ( wordPos , ( v o i d *) 1 , w o r d L e n ) ; */ } /* * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * */ // F o l l o w i n g f u n c t i o n s are o p t i o n a l . /* * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * */ // S p e c i f y a mmr s p l i t t e r f u n c t i o n if the d e f a u l t mmr s p l i t t e r is i n s u f f i c i e n t . By d e f a u l t d a t a is s p l i t t e d by u s e r d e f i n e d u n i t s i z e . // If s p l i t t e r () is defined , m m r S p l i t t e r () m u s t be d e f i n e d in a s i m i l a r way . int m m r S p l i t t e r ( m m r _ d a t a _ t * m m r D a t a ) { a s s e r t ( m m r D a t a ) ; a s s e r t ( mmrData - > f d a t a ) ; a s s e r t ( mmrData - > fpos > = 0 ) ; a s s e r t ( mmrData - > c h u n k S i z e >0) ; l o n g e n d P o s = mmrData - > f p o s + mmrData - > c h u n k S i z e -1; // R e c a l c u l a t e f p o s b a s e d on the v a l u e c a l c u l a t e d by a d e f a u l t MMR s p l i t t e r f u n c t i o n .
/* F o l l o w i n g is a w o r d c o u n t e x a m p l e w h i c h a d j u s t s f p o s so t h a t no w o r d s are cut in the m i d d l e .
if ( mmrData - > f p o s > 0) {
w h i l e ( mmrData - > f d a t a [ mmrData - > f p o s ] != ’ ’ && mmrData - > f d a t a [ mmrData - > f p o s ] != ’\ t ’
&& mmrData - > f d a t a [ mmrData - > f p o s ] != ’\ r ’ && mmrData - > f d a t a [ mmrData - > f p o s ] != ’\ n ’) { mmrData - > f p o s ++; } } */ // R e c a l c u l a t e e n d P o s b a s e d on the v a l u e c a l c u l a t e d by a d e f a u l t MMR s p l i t t e r f u n c t i o n . /* F o l l o w i n g is a w o r d c o u n t e x a m p l e w h i c h a d j u s t s e n d P o s so t h a t no w o r d s are cut in the m i d d l e .
if ( e n d P o s < mmrData - > f D a t a L e n -1) {
w h i l e ( mmrData - > f d a t a [ e n d P o s +1] != ’ ’ && mmrData - > f d a t a [ e n d P o s +1] != ’\ t ’ && mmrData - > f d a t a [ e n d P o s +1] != ’\ r ’ && mmrData - > f d a t a [ e n d P o s +1] != ’\ n ’) { e n d P o s ++; } } */ // R e c a l c u l a t e c h u n k S i z e .
mmrData - > c h u n k S i z e = endPos - mmrData - > f p o s +1; r e t u r n 0;
/* * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * */
// S p e c i f y the key / v a l u e p a c k i n g m e t h o d for MPI c o m m u n i c a t i o n . It is c a l l e d if r e s u l t f r o m e a c h n o d e s h o u l d be s e n t to n o d e 0 for r e d u c t i o n .
int m m r P a c k T y p e s ( c o n s t v o i d * key , c o n s t v o i d * val , int * k e y V a l L e n , v o i d ** k e y V a l ) { /* F o l l o w i n g is a w o r d c o u n t e x a m p l e in w h i c h key t y p e is s t r i n g and v a l u e t y p e is int . c h a r * k e y S t r = *( c h a r **) key ; k e y V a l L e n [0] = s t r l e n ( k e y S t r ) +1; k e y V a l L e n [1] = 1; k e y V a l [0] = k e y S t r ; k e y V a l [1] = ( int *) val ; */ r e t u r n 0; } /* * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * */
// S p e c i f y the key / v a l u e u n p a c k i n g m e t h o d for MPI c o m m u n i c a t i o n . It is c a l l e d if n o d e 0 s h o u l d r e c e i v e r e s u l t f r o m e a c h n o d e for r e d u c t i o n .
int m m r U n p a c k T y p e s ( v o i d ** key , v o i d ** val , c o n s t int * k e y V a l L e n , v o i d ** k e y V a l ) {
/* F o l l o w i n g is a w o r d c o u n t e x a m p l e in w h i c h key t y p e is s t r i n g and v a l u e t y p e is int .
* key = ( c h a r *) k e y V a l [ 0 ] ;
*( int *) val = *( int *) k e y V a l [ 1 ] ; */
r e t u r n 0; }
// S p e c i f y a mmr r e d u c e f u n c t i o n o n l y w h e n i d e n t i t y r e d u c e is i n s u f f i c i e n t .
// If r e d u c e () is defined , m m r R e d u c e () m u s t be d e f i n e d in a s i m i l a r way .
int m m r R e d u c e ( c o n s t v o i d * val1 , c o n s t v o i d * val2 , v o i d ** val ) {
/* F o l l o w i n g is a w o r d c o u n t e x a m p l e in w h i c h v a l u e s i . e . c o u n t s are s u m m e d up for a s a m e key i . e . w o r d .
*( int *) val = ( int ) v a l 1 + ( int ) v a l 2 ; */
r e t u r n 0; }
/* * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * */
// S p e c i f y a key / v a l u e c o m p a r e f u n c t i o n if the f i n a l key / v a l u e p a i r s s h o u l d be s o r t e d . S o r t i n g w i l l be d o n e a u t o m a t i c a l l y . int m m r K e y V a l C m p ( c o n s t v o i d * v1 , c o n s t v o i d * v2 ) { /* F o l l o w i n g is a w o r d c o u n t e x a m p l e in w h i c h k e y s i . e . w o r d s are s o r t e d in a l p h a b e t i c o r d e r . S a m e k e y s are s o r t e d by v a l u e s . k e y v a l _ t * kv1 = ( k e y v a l _ t *) v1 ; k e y v a l _ t * kv2 = ( k e y v a l _ t *) v2 ; int i1 = ( int ) kv1 - > val ;
int i2 = ( int ) kv2 - > val ;
int ret = s t r c m p (( c h a r *) kv1 - > key , ( c h a r *) kv2 - > key ) ; if ( ret )
r e t u r n ret ; e l s e {
if ( i1 < i2 )
e l s e if ( i1 > i2 ) r e t u r n -1; e l s e r e t u r n 0; } */ r e t u r n 0; } /* * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * */ // S p e c i f y a s p l i t t e r f u n c t i o n if the d e f a u l t is i n s u f f i c i e n t . By d e f a u l t d a t a is s p l i t t e d by u s e r d e f i n e d u n i t s i z e .