XRootD
XrdWait41.cc
Go to the documentation of this file.
1 /******************************************************************************/
2 /* */
3 /* X r d W a i t 4 1 . c c */
4 /* */
5 /* (c) 2009 by the Board of Trustees of the Leland Stanford, Jr., University */
6 /* Produced by Andrew Hanushevsky for Stanford University under contract */
7 /* DE-AC02-76-SFO0515 with the Department of Energy */
8 /* */
9 /* This file is part of the XRootD software suite. */
10 /* */
11 /* XRootD is free software: you can redistribute it and/or modify it under */
12 /* the terms of the GNU Lesser General Public License as published by the */
13 /* Free Software Foundation, either version 3 of the License, or (at your */
14 /* option) any later version. */
15 /* */
16 /* XRootD is distributed in the hope that it will be useful, but WITHOUT */
17 /* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or */
18 /* FITNESS FOR A PARTICULAR PURPOSE. See the GNU Lesser General Public */
19 /* License for more details. */
20 /* */
21 /* You should have received a copy of the GNU Lesser General Public License */
22 /* along with XRootD in a file called COPYING.LESSER (LGPL license) and file */
23 /* COPYING (GPL license). If not, see <http://www.gnu.org/licenses/>. */
24 /* */
25 /* The copyright holder's institutional names and contributor's names may not */
26 /* be used to endorse or promote products derived from this software without */
27 /* specific prior written permission of the institution or contributor. */
28 /******************************************************************************/
29 
30 
31 
32 /* This unitily waits for the first of n file locks. The syntax is:
33 
34  wait41 <path> [<path> [. . .]]
35 
36 */
37 
38 /******************************************************************************/
39 /* i n c l u d e f i l e s */
40 /******************************************************************************/
41 
42 #include <cctype>
43 #include <dirent.h>
44 #include <fcntl.h>
45 #include <signal.h>
46 #include <cstdlib>
47 #include <cstdio>
48 #include <cstring>
49 #include <strings.h>
50 #include <unistd.h>
51 #include <sys/param.h>
52 #include <sys/stat.h>
53 #include <sys/types.h>
54 
55 #include "XrdOuc/XrdOucTList.hh"
56 #include "XrdSys/XrdSysE2T.hh"
57 #include "XrdSys/XrdSysHeaders.hh"
58 #include "XrdSys/XrdSysPlatform.hh"
59 #include "XrdSys/XrdSysPthread.hh"
60 
61 /******************************************************************************/
62 /* L o c a l C l a s s e s */
63 /******************************************************************************/
64 
66 {
67 public:
68 
69 static void Serialize(XrdOucTList *gfP, int Wait=1);
70 
71 static int Wait41(XrdOucTList *fP);
72 
75 
76 private:
77 static XrdSysMutex gateMutex;
78 static XrdSysSemaphore gateSem;
79 static int gateOpen;
80 };
81 
82 XrdSysMutex XrdW41Gate::gateMutex;
83 XrdSysSemaphore XrdW41Gate::gateSem(0);
84 int XrdW41Gate::gateOpen = 0;
85 
87 {
88 public:
89 
90 static XrdOucTList *Expand(const char *Path, XrdOucTList *ptl);
91 };
92 
93 /******************************************************************************/
94 /* E x t e r n a l T h r e a d I n t e r f a c e s */
95 /******************************************************************************/
96 
97 namespace XrdWait41
98 {
99 void *GateWait(void *parg)
100 {
101  XrdOucTList *fP = (XrdOucTList *)parg;
102 
103 // Serialize
104 //
106  return (void *)0;
107 }
108 }
109 
110 using namespace XrdWait41;
111 
112 /******************************************************************************/
113 /* m a i n */
114 /******************************************************************************/
115 
116 int main(int argc, char *argv[])
117 {
118  sigset_t myset;
119  XrdOucTList *gateFiles = 0;
120  struct stat Stat;
121  const char *eText;
122  char buff[8];
123  int i;
124 
125 // Turn off sigpipe and host a variety of others before we start any threads
126 //
127  signal(SIGPIPE, SIG_IGN); // Solaris optimization
128  sigemptyset(&myset);
129  sigaddset(&myset, SIGPIPE);
130  sigaddset(&myset, SIGCHLD);
131  pthread_sigmask(SIG_BLOCK, &myset, NULL);
132 
133 // Set the default stack size here
134 //
135  if (sizeof(long) > 4) XrdSysThread::setStackSize((size_t)1048576);
136  else XrdSysThread::setStackSize((size_t)786432);
137 
138 // Construct a list of files. For each directory, expand that to a list
139 //
140  for (i = 1; i < argc; i++)
141  {if (stat(argv[i], &Stat))
142  {eText = XrdSysE2T(errno);
143  std::cerr <<"wait41: " <<eText <<" processing " <<argv[i] <<std::endl;
144  continue;
145  }
146  if (S_ISREG(Stat.st_mode))
147  gateFiles = new XrdOucTList(argv[i],0,gateFiles);
148  else if (S_ISDIR(Stat.st_mode))
149  gateFiles = XrdW41Dirs::Expand(argv[i], gateFiles);
150  }
151 
152 // If we have no waiters then fail
153 //
154  if (!gateFiles)
155  {std::cerr <<"wait41: Nothing to wait on!" <<std::endl;
156  std::cout <<"BAD\n" <<std::endl;
157  _exit(1);
158  }
159 
160 // Now wait for the first lock
161 //
162  eText = (XrdW41Gate::Wait41(gateFiles) ? "OK\n" : "BAD\n");
163  std::cout <<eText <<std::endl;
164 
165 // Now wait for the process to die
166 //
167  if (read(STDIN_FILENO, buff, sizeof(buff))) {}
168  exit(0);
169 }
170 
171 /******************************************************************************/
172 /* C l a s s X r d W 4 1 D i r s I m p l e m e n t a t i o n */
173 /******************************************************************************/
174 /******************************************************************************/
175 /* E x p a n d */
176 /******************************************************************************/
177 
179 {
180  struct dirent *dp;
181  struct stat Stat;
182  const char *eText;
183  char buff[1024], *sfxDir;
184  DIR *DFD;
185 
186  if (!(DFD = opendir(Path)))
187  {eText = XrdSysE2T(errno);
188  std::cerr <<"wait41: " <<eText <<" opening directory" <<Path <<std::endl;
189  return ptl;
190  }
191 
192  strcpy(buff, Path); sfxDir = buff + strlen(Path);
193  if (*(sfxDir-1) != '/') *sfxDir++ = '/';
194 
195  errno = 0;
196  while((dp = readdir(DFD)))
197  {if (!strcmp(dp->d_name, ".") || !strcmp(dp->d_name, "..")) continue;
198  strcpy(sfxDir, dp->d_name);
199  if (stat(buff, &Stat))
200  {eText = XrdSysE2T(errno);
201  std::cerr <<"wait41: " <<eText <<" processing " <<buff <<std::endl;
202  continue;
203  }
204  if (S_ISREG(Stat.st_mode)) ptl = new XrdOucTList(buff, 0, ptl);
205  errno = 0;
206  }
207 
208  if (errno)
209  {eText = XrdSysE2T(errno);
210  std::cerr <<"wait41: " <<eText <<" reading directory" <<Path <<std::endl;
211  }
212 
213  closedir(DFD);
214  return ptl;
215 }
216 
217 /******************************************************************************/
218 /* C l a s s X r d W 4 1 G a t e I m p l e m e n t a t i o n */
219 /******************************************************************************/
220 /******************************************************************************/
221 /* S e r i a l i z e */
222 /******************************************************************************/
223 
224 void XrdW41Gate::Serialize(XrdOucTList *gfP, int Wait)
225 {
226  FLOCK_t lock_args;
227  int Act, rc;
228 
229 // Establish locking options
230 //
231  bzero(&lock_args, sizeof(lock_args));
232  lock_args.l_type = F_WRLCK;
233  Act = (Wait ? F_SETLKW : F_SETLK);
234 
235 // Now perform the action
236 //
237  do {rc = fcntl(gfP->val, Act, &lock_args);} while(rc == -1 && errno == EINTR);
238 
239 // Determine result
240 //
241  if (rc != -1) rc = 0;
242  else {rc = errno;
243  std::cerr <<"Serialize: " <<XrdSysE2T(rc) <<" locking FD " <<gfP->text <<std::endl;
244  }
245 
246 // Reflect what happened here
247 //
248  gateMutex.Lock();
249  if (rc || gateOpen) close(gfP->val);
250  else gateOpen = 1;
251  gateSem.Post();
252  gateMutex.UnLock();
253 }
254 
255 /******************************************************************************/
256 /* W a i t 4 1 */
257 /******************************************************************************/
258 
260 {
261  static const int AMode = S_IRUSR | S_IWUSR | S_IRGRP | S_IROTH;
262  pthread_t tid;
263  const char *eTxt;
264  int rc, Num = 0;
265 
266 // Run through the chain of files setting up a wait. We try to do a fast
267 // redispatch in case we get a lock early.
268 //
269  while(gfP)
270  {if (Num)
271  {gateMutex.Lock();
272  if (gateOpen) {gateMutex.UnLock(); return 1;}
273  gateMutex.UnLock();
274  }
275  if ((gfP->val = open(gfP->text, O_CREAT|O_RDWR, AMode)) < 0)
276  {eTxt = XrdSysE2T(errno);
277  std::cerr <<"Wait41: " <<eTxt <<" opening " <<gfP->text <<std::endl;
278  }
279  else if ((rc = XrdSysThread::Run(&tid, GateWait, (void *)gfP,
280  XRDSYSTHREAD_BIND, "Gate Wait")))
281  {eTxt = XrdSysE2T(errno);
282  std::cerr <<"Wait41: " <<eTxt <<" creating gate thread for "
283  <<gfP->text <<std::endl;
284  close(gfP->val);
285  } else Num++;
286  gfP = gfP->next;
287  }
288 
289 // At this point we will have to wait for the lock if we have any threads
290 //
291  while(Num--)
292  {gateSem.Wait();
293  gateMutex.Lock();
294  if (gateOpen) {gateMutex.UnLock(); return 1;}
295  gateMutex.UnLock();
296  }
297 
298 // No such luck, every thread failed
299 //
300  return 0;
301 }
struct stat Stat
Definition: XrdCks.cc:49
int stat(const char *path, struct stat *buf)
struct dirent * readdir(DIR *dirp)
int open(const char *path, int oflag,...)
int fcntl(int fd, int cmd,...)
int closedir(DIR *dirp)
ssize_t read(int fildes, void *buf, size_t nbyte)
DIR * opendir(const char *path)
#define close(a)
Definition: XrdPosix.hh:43
XrdOucString Path
const char * XrdSysE2T(int errcode)
Definition: XrdSysE2T.cc:104
#define FLOCK_t
#define XRDSYSTHREAD_BIND
int main(int argc, char *argv[])
Definition: XrdWait41.cc:116
char * text
Definition: XrdOucTList.hh:46
static int Run(pthread_t *, void *(*proc)(void *), void *arg, int opts=0, const char *desc=0)
static void setStackSize(size_t stsz, bool force=false)
static XrdOucTList * Expand(const char *Path, XrdOucTList *ptl)
Definition: XrdWait41.cc:178
static int Wait41(XrdOucTList *fP)
Definition: XrdWait41.cc:259
static void Serialize(XrdOucTList *gfP, int Wait=1)
Definition: XrdWait41.cc:224
void * GateWait(void *parg)
Definition: XrdWait41.cc:99