view release on metacpan or  search on metacpan

Native.xs  view on Meta::CPAN

        res->type = type;
        res->arg = NULL;
        res->dequeued = 0;
        
        DNS_thread_arg *arg = malloc(sizeof(DNS_thread_arg));
        arg->self = self;
        arg->host = strlen(host) ? savepv(host) : NULL;
        arg->service = strlen(service) ? savepv(service) : NULL;
        arg->hints = hints;
        arg->extra = 0;
        arg->queued  = 0;
        arg->res = res;
        
        pthread_mutex_lock(&self->mutex);
        DNS_free_timedout(self, 0);
        bstree_put(self->fd_map, fd[0], res);
        if (self->pool) {
            if (self->busy_threads == self->pool && (self->extra_thread || queue_size(self->tout_queue) > self->extra_threads_cnt)) {
                arg->extra = 1;
                self->extra_threads_cnt++;
            }
            else {
                arg->queued = 1;
                queue_push(self->in_queue, arg);
                sem_post(&self->semaphore);
            }
        }
        pthread_mutex_unlock(&self->mutex);
        
        if (!self->pool || arg->extra) {
            pthread_t tid;
            
            pthread_mutex_lock(&self->mutex);
            int rc = pthread_create(&tid, &self->thread_attrs, DNS_getaddrinfo, (void *)arg);
            if (rc == 0) {
                ++self->active_threads_cnt;
                pthread_mutex_unlock(&self->mutex);
            }
            else {
                pthread_mutex_unlock(&self->mutex);
                if (arg->host)    Safefree(arg->host);
                if (arg->service) Safefree(arg->service);
                free(arg);
                free(res);
                if (hints) free(hints);
                pthread_mutex_lock(&self->mutex);
                bstree_del(self->fd_map, fd[0]);
                pthread_mutex_unlock(&self->mutex);
                close(fd[0]);
                close(fd[1]);
                croak("pthread_create(): %s", strerror(rc));
            }
        }
        
        RETVAL = fd[0];
    OUTPUT:
        RETVAL

void
_get_result(Net_DNS_Native *self, int fd)
    PPCODE:
        pthread_mutex_lock(&self->mutex);
        DNS_result *res = bstree_get(self->fd_map, fd);
        bstree_del(self->fd_map, fd);
        pthread_mutex_unlock(&self->mutex);
        
        if (res == NULL) croak("attempt to get result which doesn't exists");
        if (!res->arg) {
            pthread_mutex_lock(&self->mutex);
            bstree_put(self->fd_map, fd, res);
            pthread_mutex_unlock(&self->mutex);
            croak("attempt to get not ready result");
        }
        
        XPUSHs(sv_2mortal(newSViv(res->type)));
        SV *err = newSV(0);
        sv_setiv(err, (IV)res->gai_error);
        sv_setpv(err, res->gai_error ? gai_strerror(res->gai_error) : "");
        if (res->gai_error == EAI_SYSTEM)
            sv_catpvf(err, " (%s)", strerror(res->sys_error));
        SvIOK_on(err);
        XPUSHs(sv_2mortal(err));
        
        if (!res->gai_error) {
            struct addrinfo *info;
            for (info = res->hostinfo; info != NULL; info = info->ai_next) {
                HV *hv_info = newHV();
                hv_store(hv_info, "family", 6, newSViv(info->ai_family), 0);
                hv_store(hv_info, "socktype", 8, newSViv(info->ai_socktype), 0);
                hv_store(hv_info, "protocol", 8, newSViv(info->ai_protocol), 0);
                hv_store(hv_info, "addr", 4, newSVpvn((char*)info->ai_addr, info->ai_addrlen), 0);
                hv_store(hv_info, "canonname", 9, info->ai_canonname ? newSVpv(info->ai_canonname, 0) : newSV(0), 0);
                XPUSHs(sv_2mortal(newRV_noinc((SV*)hv_info)));
            }
            
            if (res->hostinfo) freeaddrinfo(res->hostinfo);
        }
        
        close(fd);
        close(res->fd1);
        if (res->arg->hints)   free(res->arg->hints);
        if (res->arg->host)    Safefree(res->arg->host);
        if (res->arg->service) Safefree(res->arg->service);
        free(res->arg);
        free(res);

void
_timedout(Net_DNS_Native *self, int fd)
    PPCODE:
        char unknown = 0;
        
        pthread_mutex_lock(&self->mutex);
        if (bstree_get(self->fd_map, fd) == NULL) {
            unknown = 1;
        }
        else {
            queue_push(self->tout_queue, (void*)(intptr_t)fd);
        }
        pthread_mutex_unlock(&self->mutex);
        
        if (unknown)
            croak("attempt to set timeout on unknown source");

void
DESTROY(Net_DNS_Native *self)
    CODE:
        if (PERL_GET_THX != self->perl) {
            // attempt to destroy from another perl thread
            return;
        }
        
        if (self->pool) {
            pthread_mutex_lock(&self->mutex);
            if (queue_size(self->in_queue) > 0) {
                // warnings are useless in global destruction
                if (!PL_dirty)
                    warn("destroying Net::DNS::Native object while queue for resolver has %d elements", queue_size(self->in_queue));
                
                queue_iterator *it = queue_iterator_new(self->in_queue);
                DNS_thread_arg *arg;
                
                while (!queue_iterator_end(it)) {
                    arg = queue_at(self->in_queue, it);
                    arg->res->dequeued = 1;
                    free(arg);
                    queue_iterator_next(it);
                }
                
                queue_iterator_destroy(it);
                queue_clear(self->in_queue);
            }
            pthread_mutex_unlock(&self->mutex);
            
            int i;
            for (i=0; i<self->pool; i++) {
                sem_post(&self->semaphore);
            }
        }
        
        // wait all active threads to be finished
        pthread_mutex_lock(&self->mutex);
        while (self->active_threads_cnt != 0) {
            pthread_cond_wait(&self->cv, &self->mutex);
        }
        pthread_mutex_unlock(&self->mutex);
        
        DNS_free_timedout(self, 0);
        
        if (bstree_size(self->fd_map) > 0) {
            if (!PL_dirty)
                warn("destroying Net::DNS::Native object with %d non-received results", bstree_size(self->fd_map));
            
            int *fds = bstree_keys(self->fd_map);
            int i, l;
            
            for (i=0, l=bstree_size(self->fd_map); i<l; i++) {
                DNS_result *res = bstree_get(self->fd_map, fds[i]);
                
                if (!res->dequeued) {
                    if (!res->gai_error && res->hostinfo) freeaddrinfo(res->hostinfo);
                    if (res->arg->hints)   free(res->arg->hints);
                    if (res->arg->host)    Safefree(res->arg->host);
                    if (res->arg->service) Safefree(res->arg->service);
                    free(res->arg);
                }
                
                close(res->fd1);
                close(fds[i]);
                free(res);
            }
            
            free(fds);
        }
        
        queue_iterator *it = queue_iterator_new(DNS_instances);
        while (!queue_iterator_end(it)) {
            if (queue_at(DNS_instances, it) == self) {
                queue_del(DNS_instances, it);
                break;
            }
            queue_iterator_next(it);
        }
        queue_iterator_destroy(it);
        
        if (self->in_queue) {
            queue_destroy(self->in_queue);
            sem_destroy(&self->semaphore);
        }
        pthread_attr_destroy(&self->thread_attrs);
        pthread_mutex_destroy(&self->mutex);
        pthread_cond_destroy(&self->cv);
        bstree_destroy(self->fd_map);
        queue_destroy(self->tout_queue);
        Safefree(self);

void
pack_sockaddr_in6(int port, SV *sv_address)
    PPCODE:
        STRLEN len;
        char *address = SvPV(sv_address, len);
        if (len != 16)
            croak("address length is %lu should be 16", len);
        
        struct sockaddr_in6 *addr = malloc(sizeof(struct sockaddr_in6));
        memcpy(addr->sin6_addr.s6_addr, address, 16);
        addr->sin6_family = AF_INET6;
        addr->sin6_port = port;
        
        XPUSHs(sv_2mortal(newSVpvn((char*) addr, sizeof(struct sockaddr_in6))));

void
unpack_sockaddr_in6(SV *sv_addr)
    PPCODE:
        STRLEN len;
        char *addr = SvPV(sv_addr, len);
        if (len != sizeof(struct sockaddr_in6))
            croak("address length is %lu should be %lu", len, sizeof(struct sockaddr_in6));
        
        struct sockaddr_in6 *struct_addr = (struct sockaddr_in6*) addr;
        XPUSHs(sv_2mortal(newSViv(struct_addr->sin6_port)));
        XPUSHs(sv_2mortal(newSVpvn((char*)struct_addr->sin6_addr.s6_addr, 16)));

int
_is_non_safe_symbols_loaded()
    INIT:
        char found = 0;
    CODE:
#ifdef __linux__
        dl_iterate_phdr(_dl_phdr_cb, (void*)&found);
#endif
        RETVAL = found;
    OUTPUT:
        RETVAL