#include <pwd.h>
#include <sys/types.h>
#include <unistd.h>
#include <errno.h>
#include <stdio.h>
#include <sys/times.h>
#include <stdlib.h>
#include <sys/wait.h>
#include "ace/SOCK_Stream.h"
#include "ace/INET_Addr.h"
#include "ace/OS.h"
#include "ace/SOCK_Dgram.h"
#include "ace/INET_Addr.h"
#include "ace/Reactor.h"
#include "ace/Synch.h"
#include "ace/OS.h"
#include "ace/INET_Addr.h"
#include "ace/Acceptor.h"
#include "ace/Connector.h"
#include "ace/SOCK_Acceptor.h"
#include "ace/SOCK_Connector.h"
#include "ace/Reactor.h"
#include "ace/SOCK_Stream.h"
#include "ace/SOCK_Dgram.h"
#include "ace/Log_Msg.h"
#include "ace/Log_Priority.h"
#include "ace/Get_Opt.h"
#include "time.h"
#include "pexec_share.h"
#include "wd.h"
// global variables
//struct buff_stdout outbuff;
//struct buff_stderr errbuff;
//int exitcode;
extern int errno;
enum status_type execprocess(char* uname,
char* cmdline,
int nmili_sec,
int& exitcode,
struct buff_stdout& outbuff,
struct buff_stderr& errbuff);
class PEXEC_Handler:
public ACE_Svc_Handler<ACE_SOCK_STREAM,ACE_MT_SYNCH>
{
public:
PEXEC_Handler(){
}
int init();
int open(void*);
int handle_input(ACE_HANDLE);
int handle_timeout(const ACE_Time_Value&, const void*);
int handle_close(ACE_HANDLE,ACE_Reactor_Mask);
int svc();
private:
param_pack pex_pps;
ACE_Thread_Mutex mutex_;
};
typedef ACE_Acceptor<PEXEC_Handler, ACE_SOCK_ACCEPTOR> PEXEC_Acceptor;
int
PEXEC_Handler::open(void *)
{
//ACE_DEBUG((LM_DEBUG,"the gsd_handler is going to be opened\n"));
ACE_Reactor::instance()->register_handler(this,ACE_Event_Handler::READ_MASK);
return 0;
}
int
PEXEC_Handler::handle_input(ACE_HANDLE)
{
// int rlen = peer().recv(tmp_recv,128,0);
struct param_pack pps, net_pps;
//enum status_type exec_status;
//int nint;
int rlen = peer().recv((char *)&net_pps, sizeof(net_pps), 0);
if (rlen <= -1) // can not recv messge from client
{
cout << "[" << __FILE__ << ":" << __LINE__ << "]" << "\t";
/*new add 0525*/
peer().close();
/*end add*/
ACE_ERROR_RETURN((LM_ERROR,"(%P|%t, %p)","recv message failed"),-1);
return -1;
}
if (rlen == 0){
peer().close();
return -1;
}
//convert net_pps to pps
strncpy(pps.user, net_pps.user, sizeof(pps.user));
strncpy(pps.cmdline, net_pps.cmdline, sizeof(pps.cmdline));
pps.fexitcode = ntohl(net_pps.fexitcode);
pps.milisec = ntohl(net_pps.milisec);
//cout<<"user name is "<<pps.user<<endl
// <<"cmdline is "<<pps.cmdline<<endl
//<<"exitcode only is "<<pps.exitcode_only<<endl
//<<"timeout is "<<pps.timeout<<endl;
mutex_.acquire();
pex_pps = net_pps;
activate(THR_NEW_LWP,
1, //2 new threads
1, //force active false, if already created don~Rt try again.
ACE_DEFAULT_THREAD_PRIORITY,//Use default thread priority
-1,
this,//Which ACE_Task object to create? In this case this one.
0,// don~Rt care about thread handles used
0,// don~Rt care about where stacks are created
0);//don~Rt care about stack sizes
//thread_names); // keep identifiers in thread_names
return 0;
}
int
PEXEC_Handler::handle_close(ACE_HANDLE,ACE_Reactor_Mask)
{
return 0;
}
int
PEXEC_Handler::handle_timeout(const ACE_Time_Value&, const void*) {
send_wd_heartbeat(PEXEC_SRC_TYPE);
return 0;
}
int
PEXEC_Handler::svc() {
struct buff_stdout outbuff;
struct buff_stderr errbuff;
int exitcode;
status_type exec_status;
int nint;
param_pack pps;
pps = pex_pps;
mutex_.release();
//exec cmdline
exec_status = execprocess(pps.user,
pps.cmdline,
pps.milisec,
exitcode,
outbuff,
errbuff);
nint = htonl(exec_status);
if (peer().send_n((char *)&nint, sizeof(nint), 0) == -1){
ACE_DEBUG((LM_DEBUG, ACE_TEXT("send data error : %s"),
ACE_OS::strerror(ACE_OS::last_error())));
return -1;
}
//cout<<"errbuff.len = "<<errbuff.len<<endl;
//cout<<errbuff.buffer<<endl;
if (exec_status == EXEC_SUCCESS) {
//send exit code back
nint = htonl(exitcode);
if (peer().send_n((char *)&nint, sizeof(nint)) == -1) {
ACE_DEBUG((LM_DEBUG, ACE_TEXT("send data error : %s"),
ACE_OS::strerror(ACE_OS::last_error())));
//ACE_OS::printf("send data failed\n");
return -1;
}
if (pps.fexitcode) {
//send stdout len back
nint = htonl(outbuff.len);
if (peer().send_n((char *)&nint, sizeof(nint), 0) == -1) {
//ACE_OS::printf("send data failed\n");
ACE_DEBUG((LM_DEBUG, ACE_TEXT("send data error : %s"),
ACE_OS::strerror(ACE_OS::last_error())));
return -1;
}
if (outbuff.len > 0) {
if (peer().send_n(outbuff.buffer, outbuff.len, 0) == -1) {
//ACE_OS::printf("send data failed\n");
ACE_DEBUG((LM_DEBUG, ACE_TEXT("send data error : %s"),
ACE_OS::strerror(ACE_OS::last_error())));
return -1;
}
}
//send stderr len back
nint = htonl(errbuff.len);
if (peer().send_n((char *)&nint, sizeof(nint), 0) == -1) {
//ACE_OS::printf("send data failed\n");
ACE_DEBUG((LM_DEBUG, ACE_TEXT("send data error : %s"),
ACE_OS::strerror(ACE_OS::last_error())));
return -1;
}
if (errbuff.len > 0) {
if (peer().send_n(errbuff.buffer, errbuff.len, 0) == -1) {
//ACE_OS::printf("send data failed\n");
ACE_DEBUG((LM_DEBUG, ACE_TEXT("send data error : %s"),
ACE_OS::strerror(ACE_OS::last_error())));
return -1;
}
}
}
}
// peer().close();
return 0;
}
int main(int argc, char** argv)
{
ACE_INET_Addr tcp_addr(PEXEC_PORT);
ACE_INET_Addr remote_addr;
PEXEC_Acceptor *acceptor;
ACE::daemonize();
//open log
ofstream *output_file = new ofstream("/home/dware/pexecd.log", ios::out);
if (output_file && output_file->rdstate() == ios::goodbit)
ACE_LOG_MSG->msg_ostream(output_file, 1);
ACE_LOG_MSG->open(argv[0], ACE_Log_Msg::OSTREAM,0);
ACE_DEBUG ((LM_DEBUG, ACE_TEXT ("%T (%t): Starting service.\n")));
acceptor = new PEXEC_Acceptor;
if(acceptor->open(tcp_addr, ACE_Reactor::instance()) == -1)
{
char error_message[64];
printf("[%s|%d]open failed: %s", __FILE__, __LINE__, strerror(errno));
}
PEXEC_Handler My_Pexec_Handler;
ACE_Time_Value timeout(HB_TIMEOUT);
ACE_Reactor::instance()->schedule_timer(&My_Pexec_Handler,0,timeout,timeout);
while(1)
ACE_Reactor::instance()->handle_events();
return 0;
}
status_type execprocess(char* uname,
char* cmdline,
int nmili_sec,
int& exitcode,
struct buff_stdout& outbuff,
struct buff_stderr& errbuff) {
int chuser;
struct passwd *pw;
int pp_out[2], pp_err[2];
pid_t pid;
uid_t uid;
gid_t gid;
char *shellpath, *shellname;
fd_set readfd, allfd;
int nfds, rt_sel;
struct timeval tm;
int nrd_out, nrd_err;
size_t nleft_out, nleft_err;
char *ppos_out, *ppos_err;
char buff_rub[BUFF_SIZ];
int status;
int i;
//enum status_type exec_status;
chuser = strcmp(uname, "root"); /* if we are required to change user id */
//printf("chuser is %d\n", chuser);
if ((pw = getpwnam(uname)) == NULL) { /* get passwd file entry */
ACE_DEBUG((LM_DEBUG, ACE_TEXT("getpwnam error : %s"), strerror(errno)));
return ERR_EXEC_COMMAND;
}
uid = pw->pw_uid;
gid = pw->pw_gid;
//printf("%s : uid is %d, gid is %d\n", uname, uid, gid);
/* create pipes for stdout and stderr of child process */
if (pipe(pp_out) != 0) {
ACE_DEBUG((LM_DEBUG, ACE_TEXT("create stdout pipe error : %s"), strerror(errno)));
return ERR_EXEC_COMMAND;
}
if (pipe(pp_err) != 0) {
ACE_DEBUG((LM_DEBUG, ACE_TEXT("create stderr pipe error : %s"), strerror(errno)));
return ERR_EXEC_COMMAND;
}
/* create child process */
pid = fork();
if (pid < 0) {
ACE_DEBUG((LM_DEBUG, ACE_TEXT("fork error : %s"), strerror(errno)));
return ERR_EXEC_COMMAND;
}
if (pid == 0) { /* child process */
close(pp_out[0]); /* close read end of stdout pipe */
close(pp_err[0]); /* close read end of stderr pipe */
if (dup2(pp_out[1], 1) == -1) { /* make stdout a copy of write end of pipe */
ACE_DEBUG((LM_DEBUG, ACE_TEXT("dup error : %s"), strerror(errno)));
return ERR_EXEC_COMMAND;
}
close(pp_out[1]);
if (dup2(pp_err[1], 2) == -1) { /* make stderr a copy of write end of pipe */
ACE_DEBUG((LM_DEBUG, ACE_TEXT("dup error : %s"), strerror(errno)));
return ERR_EXEC_COMMAND;
}
close(pp_err[1]);
if (chuser != 0) {
//printf("now changing uid and gid\n");
uid = pw->pw_uid;
gid = pw->pw_gid;
errno = 0;
if (setgid(gid) != 0) {
ACE_DEBUG((LM_DEBUG, ACE_TEXT("setgid error : %s"), strerror(errno)));
return ERR_EXEC_COMMAND;
}
if (setuid(uid) != 0) {
ACE_DEBUG((LM_DEBUG, ACE_TEXT("setuid error : %s"), strerror(errno)));
return ERR_EXEC_COMMAND;
}
}
for (i = 3; i < 65536; i++)
ACE_OS::close(i);
//printf("now uid is %d, gid is %d\n", getuid(), getgid());
shellpath = pw->pw_shell;
shellname = strrchr(shellpath, '/');
execl(shellpath, shellname, "-c", cmdline, NULL); /* execute new process */
perror("");
exit(EXIT_FAILURE);
}
/* in parent process */
close(pp_out[1]); /* close write end of stdout pipe */
close(pp_err[1]); /* close write end of stderr pipe */
/* get num of fds to detect */
nfds = (pp_out[0] > pp_err[0]) ? pp_out[0] : pp_err[0];
nfds++;
/* initialization */
tm.tv_usec = (nmili_sec % 1000) * 1000;
tm.tv_sec = nmili_sec / 1000;
nleft_out = STDOUT_BUFF_SIZ;
nleft_err = STDERR_BUFF_SIZ;
outbuff.len = errbuff.len = 0;
ppos_out = outbuff.buffer;
ppos_err = errbuff.buffer;
nrd_out = nrd_err = 0;
FD_ZERO(&allfd);
FD_SET(pp_out[0], &allfd);
FD_SET(pp_err[0], &allfd);
while (1) {
readfd = allfd;
if (nmili_sec > 0)
rt_sel = select(nfds, &readfd, NULL, NULL, &tm);
else
rt_sel = select(nfds, &readfd, NULL, NULL, NULL);
if (rt_sel > 0) {
if (FD_ISSET(pp_out[0], &readfd) != 0) {
if (nleft_out > 0) {
nrd_out = read(pp_out[0], ppos_out, nleft_out);
if (nrd_out > 0) {
ppos_out += nrd_out;
//outbuff.len += nrd_out;
nleft_out -= nrd_out;
}
}
else
nrd_out = read(pp_out[0], buff_rub, sizeof(buff_rub));
}
if (FD_ISSET(pp_err[0], &readfd) != 0) {
if (nleft_err > 0) {
nrd_err = read(pp_err[0], ppos_err, nleft_err);
if (nrd_err > 0) {
ppos_err += nrd_err;
//errbuff.len += nrd_err;
nleft_err -= nrd_err;
}
}
else
nrd_err = read(pp_err[0], buff_rub, sizeof(buff_rub));
}
if (nrd_out == 0 && nrd_err == 0) {
outbuff.len = STDOUT_BUFF_SIZ - nleft_out;
errbuff.len = STDERR_BUFF_SIZ - nleft_err;
*ppos_out = *ppos_err = '\0';
break;
}
}
else
if (rt_sel == 0) {//time out
return ERR_EXEC_TIMEOUT;
}
else
if (rt_sel < 0) {
if (errno == EINTR) {
tm.tv_sec = nmili_sec / 1000; /* reset the timeout value */
tm.tv_usec = 0;
//printf("error\n");
continue;
}
else
break;
}
}
wait(&status);
if (WIFEXITED(status)) {
exitcode = WEXITSTATUS(status);
return EXEC_SUCCESS;
}
else
return ERR_INVALID_EXITCODE;
//turn exec_status;
}
相关资源：敏捷开发V1.0.pptx
