/*
** ATOP - System & Process Monitor
**
** The program 'atop' offers the possibility to view the activity of
** the system on system-level as well as process-level.
**
** This source-file contains functions to interface with the atopgpud
** daemon that maintains statistics about the processor and memory
** utilization of the GPUs.
** ================================================================
** Author: Gerlof Langeveld
** E-mail: gerlof.langeveld@atoptool.nl
** Initial: April/August 2018
**
** 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, 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.
*/
#include <sys/types.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <string.h>
#include <stdio.h>
#include <unistd.h>
#include <stdlib.h>
#include <malloc.h>
#include "atop.h"
#include "photosyst.h"
#include "photoproc.h"
#include "gpucom.h"
#define DUMMY ' '
#define GPUDELIM '@'
#define PIDDELIM '#'
#define GPUDPORT 59123
static void gputype_parse(char *);
static void gpustat_parse(int, char *, int,
struct pergpu *, struct gpupidstat *);
static void gpuparse(int, char *, struct pergpu *);
static void pidparse(int, char *, struct gpupidstat *);
static int rcvuntil(int, char *, int);
static int actsock = -1;
static int numgpus;
static char **gpubusid; // array with char* to busid strings
static char **gputypes; // array with char* to type strings
static char *gputasks; // array with chars with tasksupport booleans
/*
** Open TCP connection to port of atopgpud and
** obtain type information of every GPU.
**
** Return value:
** number of GPUs
*/
int
gpud_init(void)
{
struct sockaddr_in name;
socklen_t namelen = sizeof name;
char typereq[] = {'T', APIVERSION};
uint32_t prelude;
char *buf;
int version, length;
struct timeval rcvtimeout = {2, 0}; // 2 seconds
/*
** get local socket
*/
if ( (actsock = socket(AF_INET, SOCK_STREAM, 0)) == -1)
{
perror("socket creation");
return 0;
}
/*
** connect to server port
*/
memset(&name, 0, sizeof name);
name.sin_family = AF_INET;
name.sin_addr.s_addr = htonl(INADDR_LOOPBACK);
name.sin_port = htons(GPUDPORT);
if (connect(actsock, (struct sockaddr *)&name, namelen) == -1)
goto close_and_return;
/*
** set receive timeout, not to block atop forever
** in case something fails in the commmunication
*/
(void) setsockopt(actsock, SOL_SOCKET, SO_RCVTIMEO,
&rcvtimeout, sizeof rcvtimeout);
/*
** send request: GPU types
*/
if ( write(actsock, typereq, sizeof typereq) < sizeof typereq)
{
perror("send type request to atopgpud");
goto close_and_return;
}
/*
** receive response: GPU types
*/
if (rcvuntil(actsock, (char *)&prelude, sizeof prelude) == -1)
{
perror("receive prelude from atopgpud");
goto close_and_return;
}
prelude = ntohl(prelude); // big endian to native endianess
version = (prelude >> 24) & 0xff;
length = prelude & 0xffffff;
if (version != APIVERSION)
{
fprintf(stderr,
"wrong API version from atopgpud: %d %d\n",
version, APIVERSION);
goto close_and_return;
}
if (length > 8192) // sanity check
{
fprintf(stderr,
"unexpected response length atopgpud: %d\n", length);
goto close_and_return;
}
buf = malloc(length+1);
ptrverify(buf, "Malloc failed for gpu rcvbuf\n");
if ( rcvuntil(actsock, buf, length) == -1)
{
perror("receive type request from atopgpud");
goto close_and_return;
}
buf[length] = '\0';
gputype_parse(buf);
numgpus = numgpus <= MAXGPU ? numgpus : MAXGPU;
return numgpus;
close_and_return:
close(actsock);
actsock = -1;
return 0;
}
/*
** Transmit status request for all GPUs.
**
** Calling parameters:
** void
**
** Return value:
** 0 in case of failure
** 1 in case of success
*/
int
gpud_statrequest(void)
{
char statreq[] = {'S', APIVERSION};
if (actsock == -1)
return 0;
if ( write(actsock, statreq, sizeof statreq) < sizeof statreq)
{
close(actsock);
actsock = -1;
return 0;
}
return 1;
}
/*
** Receive status response for all GPUs.
**
** Calling parameters:
** *ggs pointer to allocated array of pergpu structs
** **gps pointer to pointer in which addresses to gpupidstat structs
** are returned
** can be NULL pointer is caller is not interested in proc stats
**
** Return value:
** number of gpupidstat addresses (i.e. per-process info)
** -1 in case of failure
*/
int
gpud_statresponse(int maxgpu, struct pergpu *ggs, struct gpupidstat **gps)
{
uint32_t prelude;
char *buf = NULL, *p;
int version, length;
int pids = 0;
if (actsock == -1)
return -1;
/*
** receive 4-bytes introducer:
** first byte: API version
** next three bytes: length of string that follows
*/
if ( rcvuntil(actsock, (char *)&prelude, sizeof prelude) == -1)
{
perror("receive 4-byte prelude from atopgpud");
goto close_and_return;
}
prelude = ntohl(prelude); // big endian to native endianess
version = (prelude >> 24) & 0xff;
length = prelude & 0xffffff;
if (version != APIVERSION)
{
fprintf(stderr,
"wrong API version from atopgpud: %d %d\n",
version, APIVERSION);
goto close_and_return;
}
if (length > 8192) // sanity check
{
fprintf(stderr,
"unexpected response length atopgpud: %d\n", length);
goto close_and_return;
}
buf = malloc(length+1);
ptrverify(buf, "Malloc failed for gpu rcvbuf\n");
/*
** receive statistics string
*/
if ( rcvuntil(actsock, buf, length) == -1)
{
perror("receive stats string from atopgpud");
goto close_and_return;
}
*(buf+length) = '\0';
/*
** determine number of per-process stats
** and malloc space to parse these stats
*/
for (p=buf; *p; p++)
{
if (*p == PIDDELIM)
pids++;
}
if (gps)
{
if (pids)
{
*gps = malloc(pids * sizeof(struct gpupidstat));
ptrverify(gps, "Malloc failed for gpu pidstats\n");
memset(*gps, 0, pids * sizeof(struct gpupidstat));
}
else
{
*gps = NULL;
}
}
/*
** parse stats string for per-gpu stats
*/
gpustat_parse(version, buf, maxgpu, ggs, gps ? *gps : NULL);
free(buf);
return pids;
close_and_return:
if (buf)
free(buf);
close(actsock);
actsock = -1;
return -1;
}
/*
** Receive given number of bytes from given socket
** into given buffer address
*/
static int
rcvuntil(int sock, char *buf, int size)
{
int remain = size, n;
while (remain)
{
n = read(sock, buf, remain);
if (n <= 0)
return -1;
buf += n;
remain -= n;
}
return size;
}
/*
** Parse response string from server on 'T' request
**
** Store the type, busid and tasksupport of every GPU in
** static pointer tables
*/
static void
gputype_parse(char *buf)
{
char *p, *start, **bp, **tp, *cp;
/*
** determine number of GPUs
*/
if ( sscanf(buf, "%d@", &numgpus) != 1)
{
close(actsock);
actsock = -1;
return;
}
for (p=buf; *p; p++) // search for first delimiter
{
if (*p == GPUDELIM)
{
p++;
break;
}
}
/*
** parse GPU info and build arrays of pointers to the
** busid strings, type strings and tasksupport strings.
*/
if (numgpus) // GPUs present anyhow?
{
int field;
gpubusid = bp = malloc((numgpus+1) * sizeof(char *));
gputypes = tp = malloc((numgpus+1) * sizeof(char *));
gputasks = cp = malloc((numgpus) * sizeof(char ));
ptrverify(gpubusid, "Malloc failed for gpu busids\n");
ptrverify(gputypes, "Malloc failed for gpu types\n");
ptrverify(gputasks, "Malloc failed for gpu tasksup\n");
for (field=0, start=p; ; p++)
{
if (*p == ' ' || *p == '\0' || *p == GPUDELIM)
{
switch(field)
{
case 0:
if (p-start <= MAXGPUBUS)
*bp++ = start;
else
*bp++ = p - MAXGPUBUS;
break;
case 1:
if (p-start <= MAXGPUTYPE)
*tp++ = start;
else
*tp++ = p - MAXGPUTYPE;
break;
case 2:
*cp++ = *start;
break;
}
field++;
start = p+1;
if (*p == '\0')
break;
if (*p == GPUDELIM)
field = 0;
*p = '\0';
}
}
*bp = NULL;
*tp = NULL;
}
}
/*
** Parse entire response string from server.
**
** Every series with counters on GPU level is introduced
** with a '@' delimiter.
** Every series with counters on process level is introduced
** with a '#' delimiter (last part of the GPU level data).
*/
static void
gpustat_parse(int version, char *buf, int maxgpu,
struct pergpu *gg, struct gpupidstat *gp)
{
char *p, *start, delimlast;
int gpunum = 0;
/*
** parse stats string
*/
for (p=start=buf, delimlast=DUMMY; gpunum <= maxgpu; p++)
{
char delimnow;
if (*p != '\0' && *p != GPUDELIM && *p != PIDDELIM)
continue;
/*
** next delimiter or end-of-string found
*/
delimnow = *p;
*p = 0;
switch (delimlast)
{
case DUMMY:
break;
case GPUDELIM:
gpuparse(version, start, gg);
strcpy(gg->type, gputypes[gpunum]);
strcpy(gg->busid, gpubusid[gpunum]);
gpunum++;
gg++;
break;
case PIDDELIM:
if (gp)
{
pidparse(version, start, gp);
gp->gpu.nrgpus++;
gp->gpu.gpulist = 1<<(gpunum-1);
gp++;
(gg-1)->nrprocs++;
}
}
if (delimnow == 0 || *(p+1) == 0)
break;
start = p+1;
delimlast = delimnow;
}
}
/*
** Parse GPU statistics string
*/
static void
gpuparse(int version, char *p, struct pergpu *gg)
{
switch (version)
{
case 1:
(void) sscanf(p, "%d %d %lld %lld %lld %lld %lld %lld",
&(gg->gpupercnow), &(gg->mempercnow),
&(gg->memtotnow), &(gg->memusenow),
&(gg->samples), &(gg->gpuperccum),
&(gg->memperccum), &(gg->memusecum));
gg->nrprocs = 0;
break;
}
}
/*
** Parse PID statistics string
*/
static void
pidparse(int version, char *p, struct gpupidstat *gp)
{
switch (version)
{
case 1:
(void) sscanf(p, "%c %ld %d %d %lld %lld %lld %lld",
&(gp->gpu.state), &(gp->pid),
&(gp->gpu.gpubusy), &(gp->gpu.membusy),
&(gp->gpu.timems),
&(gp->gpu.memnow), &(gp->gpu.memcum),
&(gp->gpu.sample));
break;
}
}
/*
** Merge the GPU per-process counters with the other
** per-process counters
*/
static int compgpupid(const void *, const void *);
void
gpumergeproc(struct tstat *curtpres, int ntaskpres,
struct tstat *curpexit, int nprocexit,
struct gpupidstat *gpuproc, int nrgpuproc)
{
struct gpupidstat **gpp;
int t, g, gpuleft = nrgpuproc;
/*
** make pointer list for elements in gpuproc
*/
gpp = malloc(nrgpuproc * sizeof(struct gpupidstat *));
if (!gpp)
ptrverify(gpp, "Malloc failed for process list\n");
for (g=0; g < nrgpuproc; g++)
gpp[g] = gpuproc + g;
/*
** sort the list with pointers in order of pid
*/
if (nrgpuproc > 1)
qsort(gpp, nrgpuproc, sizeof(struct gpupidstat *), compgpupid);
/*
** accumulate entries that contain stats from same PID
** on different GPUs
*/
for (g=1; g < nrgpuproc; g++)
{
if (gpp[g-1]->pid == gpp[g]->pid)
{
struct gpupidstat *p = gpp[g-1], *q = gpp[g];
p->gpu.nrgpus += q->gpu.nrgpus;
p->gpu.gpulist |= q->gpu.gpulist;
if (p->gpu.gpubusy != -1)
p->gpu.gpubusy += q->gpu.gpubusy;
if (p->gpu.membusy != -1)
p->gpu.membusy += q->gpu.membusy;
if (p->gpu.timems != -1)
p->gpu.timems += q->gpu.timems;
p->gpu.memnow += q->gpu.memnow;
p->gpu.memcum += q->gpu.memcum;
p->gpu.sample += q->gpu.sample;
if (nrgpuproc-g-1 > 0)
memmove(&(gpp[g]), &(gpp[g+1]),
(nrgpuproc-g-1) * sizeof p);
nrgpuproc--;
g--;
}
}
/*
** merge gpu stats with sorted task list of active processes
*/
for (t=g=0; t < ntaskpres && g < nrgpuproc; t++)
{
if (curtpres[t].gen.isproc)
{
if (curtpres[t].gen.pid == gpp[g]->pid)
{
curtpres[t].gpu = gpp[g]->gpu;
gpp[g++] = NULL;
if (--gpuleft == 0 || g >= nrgpuproc)
break;
}
// anyhow resync
while ( curtpres[t].gen.pid > gpp[g]->pid)
{
if (++g >= nrgpuproc)
break;
}
}
}
if (gpuleft == 0)
{
free(gpp);
return;
}
/*
** compact list with pointers to remaining pids
*/
for (g=t=0; g < nrgpuproc; g++)
{
if (gpp[g] == NULL)
{
for (; t < nrgpuproc; t++)
{
if (gpp[t])
{
gpp[g] = gpp[t];
gpp[t] = NULL;
break;
}
}
}
}
/*
** merge remaining gpu stats with task list of exited processes
*/
for (t=0; t < nprocexit && gpuleft; t++)
{
if (curpexit[t].gen.isproc)
{
for (g=0; g < gpuleft; g++)
{
if (gpp[g] == NULL)
continue;
if (curpexit[t].gen.pid == gpp[g]->pid)
{
curpexit[t].gpu = gpp[g]->gpu;
gpp[g] = NULL;
gpuleft--;
}
}
}
}
free(gpp);
}
static int
compgpupid(const void *a, const void *b)
{
return (*((struct gpupidstat **)a))->pid - (*((struct gpupidstat **)b))->pid;
}